Flume读取数据写入Hive、Mysql

淩亂°似流年 2022-11-28 13:27 242阅读 0赞

关于flume的简介以及应用在前面已经给大家详细的讲解过，今天主要就如何把flume抓取的数据写入到其它地方。  
由于时间关系在这里还是使用伪数据和前面所说的一样，主要是配置flume，先说下把读取的数据写入到Hive中。

## 读取数据写入Hive ##

先创建hive表，数据库名就叫test吧，

create table tests ( id int , msg string )
       partitioned by (continent string, country string, time string)
       clustered by (id) into 5 buckets
       stored as orc;

配置hive的flume  
先新建配置文件用于关联hive  
还是在/opt/flumeconf下创建properties文件，并添加以下配置  
cd /opt/flumeconf  
vi conf\_0817\_hive.properties

a1.sources = s1
    a1.channels = c1
    a1.sinks = k1
    
    a1.sources.s1.type=spooldir
    a1.sources.s1.spoolDir=/opt/soft/datas
    
    a1.sinks.k1.type = hive
    a1.sinks.k1.channel = c1
    a1.sinks.k1.hive.metastore = thrift:#127.0.0.1:9083
    a1.sinks.k1.hive.database = test
    a1.sinks.k1.hive.table = tests
    a1.sinks.k1.hive.partition = asia,%{ country},%y-%m-%d-%H-%M
    a1.sinks.k1.useLocalTimeStamp = false
    
    a1.sinks.k1.round = true
    a1.sinks.k1.roundValue = 10
    a1.sinks.k1.roundUnit = minute
    a1.sinks.k1.serializer = DELIMITED
    a1.sinks.k1.serializer.delimiter = "\t"
    a1.sinks.k1.serializer.serdeSeparator = '\t'
    a1.sinks.k1.serializer.fieldnames =id,msg
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    
    a1.sinks.k1.channel = c1
    a1.sources.s1.channels = c1

此 Sink 将包含分隔文本或 JSON 数据的 Event 直接流式传输到 Hive 表或  
分区上。 Event 使用 Hive 事务进行写入，一旦将一组 Event 提交给Hive，它们就会立即显示给 Hive 查询。

即将写入的目标分区既可以预先自己创建，也可以选择让 Flume 创建它们。写入的 Event 数据中的字段将映射到 Hive 表中的相应列。  
**属性解析：**

*  **type**  
    必须是hive
 *  **hive.metastore**  
    Hive metastore URI (eg thrift:\#a.b.com:9083 )
 *  **hive.database**  
    Hive 数据库名
 *  **hive.table**  
    Hive 表名
 *  **hive.partition**  
    逗号分隔的要写入的分区信息。比如 hive 表的分区是：  
    （continent: string, country :string, time : string），那么“Asia,India,2020-08-17-01-21”就表示数据会写入到：  
    continent=Asia,country=India,time=2020-08-17-01-21 这个分区。
 *  **hive.txnsPerBatchAsk**  
    默认值100，从 Flume 等客户端接收数据流会使用多次事务来操作，而不是只开启一个事务。  
    这个参数指定处理每次请求所开启的事务数量。来自同一个批次中所有事务中的数据最终都在一个文件中。  
    Flume会向每个事务中写入 batchSize 个 Event，这个参数和 batchSize 一起控制着每个文件的大小，请注意，Hive 最终会将这些文件压缩成一个更大的文件。
 *  **heartBeatInterval**  
    发送到 Hive 的连续心跳检测间隔（秒），以防止未使用的事务过期。设置为 0 表示禁用心跳。
 *  **autoCreatePartitions**  
    Flume 会自动创建必要的 Hive 分区以进行流式传输
 *  **batchSize**  
    写入一个 Hive 事务中最大的 Event 数量
 *  **maxOpenConnections**  
    允许打开的最大连接数。如果超过此数量，则关闭最近最少使用的连接。
 *  **callTimeout**  
    Hive、HDFS I/O 操作的超时时间（毫秒），比如：开启事务、写数据、提交事务、取消事务。
 *  **serializer**  
    序列化器负责解析 Event 中的字段并把它们映射到 Hive 表中的列，选择哪种序列化器取决于Event 中 的 数 据 格 式 ， 支 持 的 序 列 化 器 有 ：DELIMITED 和 JSON。  
    使用 DELIMITED 时：  
    **serializer.delimiter**：指定字段分隔符（默认逗号）  
    **serializer.fieldnames**：Hive 表中的列（必须）。  
    如‘time,ip,message’表示输入映射到 hive 表中的time，ip 和 message 列的第 1，第 3 和第 4 个字段。
 *  **round**  
    是否启用时间戳舍入机制
 *  **roundUnit**  
    舍入值的单位，可选值：second 、 minute 、 hour
 *  **roundValue**  
    舍入到小于当前时间的最高倍数（使用 roundUnit 配 置 的 单 位 ）  
    例 子 1 ： roundUnit=second ，roundValue=10，则 14:31:18 这个时间戳会被舍入到14:31:10;  
    例 子 2 ： roundUnit=second ，roundValue=30，则 14:31:18 这个时间戳会被舍入到14:31:00，14:31:42 这个时间戳会被舍入到 14:31:30;
 *  **timeZone**  
    应用于解析分区中转义序列的时区名称，比如：America/Los\_Angeles、Asia/Shanghai、Asia/Tokyo等
 *  **useLocalTimeStamp**  
    替换转义序列时是否使用本地时间戳（否则使用Event header 中的 timestamp ）

**启动flume命令：**

flume-ng agent -n a1 -c conf -f /opt/flumeconf/conf_0817_hive.properties -Dflume.root.logger=INFO,console

## 读取数据写入Mysql ##

一样先创建一张表，默认数据库名为test

create table tests(
    id int(11) not null  auto increment primary key,
    contest varchar(500) not null)

配置mysql的flume  
先新建配置文件用于关联mysql  
还是在/opt/flumeconf下创建properties文件，并添加以下配置  
cd /opt/flumeconf  
vi conf\_0817\_mysql.properties

a1.sources = s1
    a1.sinks = k1
    a1.channels = c1
    
    a1.sources.s1.type = exec
    a1.sources.s1.command = tail -F /opt/soft/datas/test.log
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    
    a1.sinks.k1.type = org.flume.mysql.sink.MysqlSink
    a1.sinks.k1.hostname=localhost
    a1.sinks.k1.port=3306
    a1.sinks.k1.databaseName=test
    a1.sinks.k1.tableName=tests
    a1.sinks.k1.user=root
    a1.sinks.k1.password=1234
    
    a1.sinks.k1.channel = c1
    a1.sources.s1.channels = c1

**启动flume命令：**

flume-ng agent -n a1 -c conf -f /opt/flumeconf/conf_0817_mysql.properties -Dflume.root.logger=INFO,console

生成数据到目标日志文件中：

for i in { 1..10};do echo "transaction$i" >> /opt/soft/datas/test.log;done;

这样数据就会写入到数据库中。