大数据基础

# .bashrc
export JAVA_HOME=/root/hadoop/jdk8u432-b06
export HADOOP_HOME=/root/hadoop/hadoop-3.2.0
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

# etc/hadoop/hadoop-env.sh
export JAVA_HOME=/root/hadoop/jdk8u432-b06
export HADOOP_LOG_DIR=/root/hadoop/hadoop_repo/logs/hadoop

<!-- etc/hadoop/core-site.xml -->
<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://debian101:9000</value>
    </property>
    <property>
    	<name>hadoop.tmp.dir</name>
    	<value>/root/hadoop/hadoop_repo</value>
    </property>
</configuration>

<!-- etc/hadoop/hdfs-site.xml -->
<configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>

<!-- etc/hadoop/mapred-site.xml -->
<configuration>
    <property>
        <name>mapreduce.framework.name</name>
        <value>yarn</value>
    </property>
</configuration>

<!-- etc/hadoop/yarn-site.xml -->
<configuration>
    <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>
    </property>
    <property>
        <name>yarn.nodemanager.env-whitelist</name>
        <value>JAVA_HOME,HADOOP_COMMON_HOME,HADOOP_HDFS_HOME,HADOOP_CONF_DIR,CLASSPATH_PREPEND_DISTCACHE,HADOOP_YARN_HOME,HADOOP_MAPRED_HOME</value>
    </property>
</configuration>

# sbin/start-dfs.sh
HDFS_DATANODE_USER=root
HDFS_DATANODE_SECURE_USER=hdfs
HDFS_NAMENODE_USER=root
HDFS_SECONDARYNAMENODE_USER=root

# sbin/start-yarn.sh
YARN_RESOURCEMANAGER_USER=root
HADOOP_SECURE_DN_USER=yarn
YARN_NODEMANAGER_USER=root

# 格式化文件系统
bin/hdfs namenode -format
# 启动
sbin/start-all.sh

# 帮助命令
bin/hdfs dfs -help

# 从本地上传文件
bin/hdfs dfs -put README.txt /

# 列出文件
bin/hdfs dfs -ls /

# 查看文件内容
bin/hdfs dfs -cat /README.txt

# 下载文件到本地
bin/hdfs dfs -get /README.txt README.txt.bak

# 创建文件夹
bin/hdfs dfs -mkdir /test

# 删除文件
bin/hdfs dfs -rm /README.txt

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.2.0</version>
    </dependency>
</dependencies>

public class HdfsOp {
    public static void main(String[] args) throws IOException, URISyntaxException, InterruptedException {
        Configuration conf = new Configuration();
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://debian101:9000"), conf, "root");

        // 上传本地文件
        String localPath1 = "C:\\Users\\xiang\\Documents\\IdeaProjects\\bigdata\\hadoop\\pom.xml";
        FSDataOutputStream outputStream = fileSystem.create(new Path("/pom.xml"));
        IOUtils.copyBytes(Files.newInputStream(Paths.get(localPath1)), outputStream, conf, true);

        // 下载文件到本地
        String localPath2 = "C:\\Users\\xiang\\Documents\\IdeaProjects\\bigdata\\hadoop\\README.txt";
        FSDataInputStream fsis = fileSystem.open(new Path("/README.txt"));
        IOUtils.copyBytes(fsis, Files.newOutputStream(Paths.get(localPath2)), conf, true);

        // 删除文件
        boolean ignored = fileSystem.delete(new Path("/README.txt"), true);
    }
}

public class WordCountJob {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(WordCountJob.class);
        // map 相关
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        // reduce 相关
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        // 指定文件路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.waitForCompletion(true);
    }

    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
        final LongWritable one = new LongWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] words = value.toString().split(" ");
            for (String word : words) {
                context.write(new Text(word), one);
            }
        }
    }

    public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long sum = 0;
            for (final LongWritable value : values) {
                sum += value.get();
            }
            context.write(key, new LongWritable(sum));
        }
    }
}

# 提交任务
bin/hadoop jar hadoop-1.0-SNAPSHOT.jar store.xianglin.mr.WordCountJob /test/hello.txt /out

# 查看结果
bin/hdfs dfs -ls /out
bin/hdfs dfs -cat /out/part-r-00000

<!-- etc/hadoop/yarn-site.xml -->
<configuration>
    <property>
        <name>yarn.resourcemanager.hostname</name>
        <value>debian101</value>
    </property>
    <property>
        <name>yarn.log-aggregation-enable</name>
        <value>true</value>
    </property>
    <property>
        <name>yarn.log.server.url</name>
        <value>http://debian101:19888/jobhistory/logs/</value>
    </property>
</configuration>

# 启动日志服务
bin/mapred --daemon start historyserver

bin/yarn logs -applicationId application_1729607544722_0001

// 1. 在代码中指定压缩格式
Configuration conf = new Configuration();
conf.setBoolean(MRJobConfig.MAP_OUTPUT_COMPRESS, true);
conf.setClass(MRJobConfig.MAP_OUTPUT_COMPRESS_CODEC, DeflateCodec.class, CompressionCodec.class);
Job job = Job.getInstance(conf);

// 2. 通过命令参数设置压缩格式
Configuration conf = new Configuration();
String[] remainingArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
Job job = Job.getInstance(conf);

bin/hadoop jar bigdata-1.0-SNAPSHOT.jar store.xianglin.hadoop.compress.DataCompress -Dmapreduce.output.fileoutputformat.compress=true -Dmapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.DeflateCodec /words.dat /words_out/non_compress

# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

# 启动 agent
bin/flume-ng agent --conf conf --conf-file conf/example.properties --name a1 -Dflume.root.logger=INFO,console

telnet 127.0.0.1 44444

2024-10-23 15:50:32,990 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 68 65 6C 6C 6F 0D                               hello. }

# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/flume/data/studentDir

# Use file channel 
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /root/flume/data/student/checkpoint
a1.channels.c1.dataDirs = root/flume/data/student/data

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://debian101:9000/flume/studentDir
a1.sinks.k1.hdfs.filePrefix = stu-
a1.sinks.k1.hdfs.writeFormat=Text
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.hdfs.rollSize=3600
a1.sinks.k1.hdfs.hdfs.rollInterval=134217728
a1.sinks.k1.hdfs.hdfs.rollCount=0

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

bin/flume-ng agent --conf conf --conf-file conf/dirpool-to-hdfs.conf --name a1 -Dflume.root.logger=INFO,console

bin/hdfs dfs -ls /flume/studentDir

# cp conf/hive-env.sh.template conf/hive-env.sh
export JAVA_HOME=/root/hadoop/jdk8u432-b06
export HADOOP_HOME=/root/hadoop/hadoop-3.2.0
export HIVE_HOME=/root/hive/apache-hive-3.1.2-bin

<!-- cp conf/hive-default.xml.template conf/hive-site.xml -->
<!-- 配置 MySQL -->
  <property>
    <name>javax.jdo.option.ConnectionURL</name>
    <value></value>
  </property>
  <property>
    <name>javax.jdo.option.ConnectionDriverName</name>
    <value></value>
  </property>
    <property>
    <name>javax.jdo.option.ConnectionUserName</name>
    <value></value>
  </property>
    <property>
    <name>javax.jdo.option.ConnectionPassword</name>
    <value></value>
  </property>

<!-- etc/hadoop/core-site.xml -->
<property>
    <name>hadoop.proxyuser.root.hosts</name>
    <value>*</value>
</property>
<property>
    <name>hadoop.proxyuser.root.groups</name>
    <value>*</value>
</property>

# 先创建数据库
bin/schematool -dbType mysql -initSchema

# 默认以 latin1 编码建表，为显示中文注释，需修改表编码
alter table COLUMNS_V2 modify column COMMENT varchar(256) character set utf8mb4;
alter table TABLE_PARAMS modify column PARAM_VALUE varchar(4000) character set utf8mb4;

alter table PARTITION_PARAMS modify column PARAM_VALUE varchar(4000) character set utf8mb4;
alter table PARTITION_KEYS modify column PKEY_COMMENT varchar(4000) character set utf8mb4;

# 启动 hive
bin/hive

# 或者使用 beeline
bin/hiveserver2
bin/beeline -u jdbc:hive2://localhost:10000 -n root

# hive 表操作
show tables;
create table t1(id int, name string);
insert into t1(id, name) values (1, 'zs');
select * from t1;
drop table t1;

<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-jdbc</artifactId>
    <version>3.1.2</version>
</dependency>

public class HiveJdbcDemo {
    public static void main(String[] args) throws SQLException {
        String hiveUrl = "jdbc:hive2://192.168.56.101:10000";
        Connection connection = DriverManager.getConnection(hiveUrl, "root", "");
        Statement statement = connection.createStatement();
        statement.execute("create table t1(id int, name string)");
        statement.execute("insert into t1(id, name) values (1, 'zs')");
        try (ResultSet resultSet = statement.executeQuery("select * from t1")) {
            while (resultSet.next()) {
                int id = resultSet.getInt("id");
                String name = resultSet.getString("name");
                System.out.println(id + " " + name);
            }
        }
        statement.execute("drop table t1");
    }
}

# 创建数据库，指定存储位置
create database mydb1;
create database mydb2 location '/user/hive/mydb2';

-- 创建表
create table t2(id int);
-- 查看表字段
desc t2;
-- 查看建表语句
show create table t2;
-- 修改表名
alter table t2 rename to t2_bak;
-- 加载数据
load data local inpath '/root/hive/hivedata/t2.data' into table t2_bak;
-- 添加字段
alter table t2_bak add columns (name string);
-- 删除表
drop table t2;

-- 建表指定列分隔符
create table t3
(
    id           int comment 'ID',
    stu_name     string comment 'name',
    stu_birthday date comment 'birthday',
    online       boolean comment 'is online'
) row format delimited fields terminated by '\t'
    lines terminated by '\n';

load data local inpath '/root/hive/hivedata/t3.data' into table t3;

-- 创建一个外部表
create external table external_table
(
    key string
) location '/data/external';
load data local inpath '/root/hive/hivedata/external_table.data' into table external_table;
select * from external_table;
drop table external_table;
-- 外部表和内部表的转换
alter table managed_table set tblproperties ('external'='true');
alter table external_table set tblproperties ('external'='false');

-- 创建分区表
create table partition_1
(
    id   int,
    name string
) partitioned by (dt string)
    row format delimited fields terminated by '\t';
-- 加载数据到分区中
load data local inpath '/root/hive/hivedata/partition_1.data' into table partition_1 partition (dt='20200101');
-- 添加/删除分区
alter table partition_1 add partition (dt='20200102');
alter table partition_1 drop partition (dt='20200102');
-- 查看分区信息
show partitions partition_1;

-- 查看内置函数
show functions ;
-- 查看函数描述信息
desc function cos;
desc function extended cos;

-- 按学科分组，取每组中分数最高的三行
select *
from (select *, row_number() over (partition by sub order by score desc ) as num from student_score) s
where s.num <= 3;

-- zs,swing
-- zs,footbal
-- zs,sing
-- zs,codeing
-- zs,swing
select * from student_favors;
-- zs,"[""swing"",""footbal"",""sing"",""codeing"",""swing""]"
select name,collect_list(favor) as favor_list from student_favors group by name ;
-- zs,"swing,footbal,sing,codeing,swing"
select name,concat_ws(',',collect_list(favor)) as favors from student_favors group by name ;
-- zs,"swing,footbal,sing,codeing"
select name,concat_ws(',',collect_set(favor)) as favors from student_favors group by name ;

-- zs,"swing,footbal,sing"
-- ls,"codeing,swing"
select * from student_favors_2;
-- zs,"[""swing"",""footbal"",""sing""]"
-- ls,"[""codeing"",""swing""]"
select name,split(favorlist,',') from student_favors_2;
-- swing
-- footbal
-- sing
-- codeing
-- swing
select explode(split(favorlist,',')) from student_favors_2;
-- zs,swing
-- zs,footbal
-- zs,sing
-- ls,codeing
-- ls,swing
select name, favor_new from student_favors_2 lateral view explode(split(favorlist,',')) table1 as favor_new;

-- 创建一个不带压缩的表
create external table stu_textfile
(
    id   int,
    name string,
    city string
) row format delimited
    fields terminated by ','
    lines terminated by '\n'
    location 'hdfs://debian101:9000/stu_textfile';

select *
from stu_textfile
limit 2;

-- 使用 deflate 压缩的表
create external table stu_textfile_deflate_compress
(
    id   int,
    name string,
    city string
) row format delimited
    fields terminated by ','
    lines terminated by '\n'
    location 'hdfs://debian101:9000/stu_textfile_deflate_compress';
-- 当前会话开启压缩选项
set hive.exec.compress.output=true;
set mapreduce.output.fileoutputformat.compress=true;
set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.DeflateCodec;
-- 导入数据
set mapreduce.job.reduces=1;
insert into stu_textfile_deflate_compress
select id, name, city
from stu_textfile
group by id, name, city;
-- 查询数据
select id,count(*) from stu_textfile_deflate_compress group by id ;

-- 新开 session，避免与上面的 set 命令混淆
-- 使用 sequencefile 存储格式
create external table stu_seqfile_none_compress
(
    id   int,
    name string,
    city string
) row format delimited
    fields terminated by ','
    lines terminated by '\n'
    stored as sequencefile
    location 'hdfs://debian101:9000/stu_seqfile_none_compress';
-- 导入数据
set mapreduce.job.reduces=1;
insert into stu_seqfile_none_compress
select id, name, city
from stu_textfile
group by id, name, city;
-- 验证是否支持拆分：支持
select id, count(*)
from stu_seqfile_none_compress
group by id;

-- 使用 deflate 压缩
-- https://cwiki.apache.org/confluence/display/Hive/CompressedStorage
create external table stu_seqfile_deflate_compress
(
    id   int,
    name string,
    city string
) row format delimited
    fields terminated by ','
    lines terminated by '\n'
    stored as sequencefile
    location 'hdfs://debian101:9000/stu_seqfile_deflate_compress';
-- 当前会话开启压缩选项
set hive.exec.compress.output=true;
set mapreduce.output.fileoutputformat.compress=true;
set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.DeflateCodec;
set io.seqfile.compression.type=BLOCK;
-- 导入数据
set mapreduce.job.reduces=1;
insert into stu_seqfile_deflate_compress
select id, name, city
from stu_textfile
group by id, name, city;

# conf/hbase-env.sh
export JAVA_HOME=/root/hadoop/jdk8u432-b06
export HADOOP_HOME=/root/hadoop/hadoop-3.2.0
export HBASE_LOG_DIR=/root/hbase/data/log

<!-- conf/hbase-site.xml -->
<property>
  <name>hbase.unsafe.stream.capability.enforce</name>
  <value>false</value>
</property>
<property>
  <name>hbase.cluster.distributed</name>
  <value>true</value>
</property>
<property>
  <name>hbase.rootdir</name>
  <value>hdfs://debian101:9000/hbase</value>
</property>

bin/start-hbase.sh

# 集群状态
status
# 版本信息
version
# 登录用户
whoami

# 创建表，指定列簇
create 'student','inof','level'
# 查看表信息
list
desc 'student'
# 修改表列簇信息
alter 'student',{NAME=>'level',VERSIONS=>'3'}
alter 'student','about'
alter 'student',{NAME=>'about',METHOD=>'delete'}
# 查看表状态
disable/enable 'student'
is_disabled/is_enabled 'student'
# 删除表
truncate 'student'
drop 'student'

# 添加数据
put 'student','jack','info:sex','man'
put 'student','jack','info:age','22'
put 'student','jack','level:class','A'
# 查看数据
scan 'student'
get 'student','jack'
get 'student','jack','info'
get 'student','jack','info:age'
# 统计总数
count 'student'
# 删除数据
delete 'student','jack','info:age'
deleteall 'student','jack'

import org.apache.hadoop.hbase.CellUtil
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.*
import org.apache.hadoop.hbase.util.Bytes

fun main() {
    // 获取 HBase 连接
    val conf = HBaseConfiguration.create()
    conf.set("hbase.rootdir", "hdfs://debian101:9000/hbase")
    conf.set("hbase.zookeeper.quorum", "debian101:2181")
    val conn = ConnectionFactory.createConnection(conf)

    val tableName = "test"

    // 表操作:
    // 创建表
    conn.admin.use { admin ->
        val familyInfo = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info"))
            .setMaxVersions(3)
            .build()
        val familyLevel = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("level"))
            .setMaxVersions(2)
            .build()
        val descriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName))
            .setColumnFamilies(listOf(familyInfo, familyLevel))
            .build()
        admin.createTable(descriptor)
    }

    dml(conn, tableName)

    // 删除表
    conn.admin.use { admin ->
        admin.disableTable(TableName.valueOf(tableName))
        admin.deleteTable(TableName.valueOf(tableName))
    }

    conn.close()
}

private fun dml(conn: Connection, tableName: String) {
    // 数据操作:
    // 新增数据
    conn.getTable(TableName.valueOf(tableName)).use { table ->
        val put = Put(Bytes.toBytes("laowang"))
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("18"))
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("man"))
        put.addColumn(Bytes.toBytes("level"), Bytes.toBytes("class"), Bytes.toBytes("A"))
        table.put(put)
    }

    // 查询数据
    conn.getTable(TableName.valueOf(tableName)).use { table ->
        val get = Get(Bytes.toBytes("laowang"))
        get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"))
        get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"))
        get.addColumn(Bytes.toBytes("level"), Bytes.toBytes("class"))

        val result = table.get(get)
        val cells = result.listCells()
        cells.forEach { cell ->
            val family = CellUtil.cloneFamily(cell)
            val column = CellUtil.cloneQualifier(cell)
            val value = CellUtil.cloneValue(cell)

            println("列簇: ${String(family)}, 列: ${String(column)}, 值: ${String(value)}")
        }

        val value = result.getValue(Bytes.toBytes("level"), Bytes.toBytes("class"))
        println("level:class: ${String(value)}")
    }

    // 查询多版本数据
    conn.getTable(TableName.valueOf(tableName)).use { table ->
        val get = Get(Bytes.toBytes("laowang"))
        // 指定读取所有版本数据，不设置时只获取最新数据
        get.readAllVersions()
        val result = table.get(get)
        result.getColumnCells(Bytes.toBytes("info"), Bytes.toBytes("age")).forEach { cell ->
            val value = CellUtil.cloneValue(cell)
            println("值为: ${String(value)}, 时间戳: ${cell.timestamp}")
        }
    }

    // 删除数据
    conn.getTable(TableName.valueOf(tableName)).use { table ->
        val delete = Delete(Bytes.toBytes("laowang"))
        // 仅删除 age 列
        delete.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"))
        table.delete(delete)
    }
}

# cp conf/spark-env.sh.template conf/spark-env.sh
export SPARK_MASTER_HOST=debian101

# cp conf/slaves.template conf/slaves
debian101

sbin/start-all.sh

./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://debian101:7077 \
  --executor-memory 2G \
  --total-executor-cores 2 \
  examples/jars/spark-examples_2.11-2.4.3.jar \
  2

# cp conf/spark-defaults.conf.template conf/spark-defaults.conf
spark.eventLog.enabled=true
spark.eventLog.compress=true
spark.eventLog.dir=hdfs://debian101:9000/tmp/logs/root/logs
spark.history.fs.logDirectory=hdfs://debian101:9000/tmp/logs/root/logs
spark.yarn.historyServer.address=http://debian101:18080

# conf/spark-env.sh
export HADOOP_CONF_DIR=/root/hadoop/hadoop-3.2.0/etc/hadoop
export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=18080 -Dspark.history.fs.logDirectory=hdfs://debian101:9000/tmp/logs/root/logs"

sbin/start-history-server.sh

Dataset<Row> dataset = sparkSession.read().format("json").load("read_path");
dataset.write().format("csv").mode(SaveMode.ErrorIfExists).save("save_path");

bin/start-cluster.sh

# 在 YARN 中创建 Flink 集群
bin/yarn-session.sh -jm 1024m -tm 1024m -d
# 提交示例任务
bin/flink run examples/batch/WordCount.jar

# 停止 YARN 中的 Flink 集群
bin/yarn application -kill application_1729685357049_0016

# -m yarn-cluster 创建临时的 Flink 集群并提交任务
bin/flink run -m yarn-cluster -yjm 1024m -ytm 1024m examples/batch/WordCount.jar