Flink消费Kafka实时写入Doris

本文模拟实际生产环境，通过FileBeat采集日志信息到Kafka，再通过Flink消费Kafka实时写入Doris。

文章目录

Filebeat采集日志到Kafka

常见的日志采集工具有以下几种：

Flume、Logstash和Filebeat

。

• Flume采用Java编写，它是一个分布式、高度可靠且高度可用的工具，旨在高效地搜集、汇总和转移大量日志数据，该工具拥有一个简洁且灵活的流数据流架构，它配备了可调节的可靠性机制、故障切换以及恢复功能，此外，Flume通过简单且可扩展的数据模型支持在线分析应用程序。
• Logstash是一个开源的日志管理和分析工具，它能够从多个数据源收集数据，对数据进行转换和清洗，并将处理后的数据传输到目标系统。
• Filebeat是一款go语言编写的日志文件收集工具，当在服务器上部署其客户端后，它会持续监听特定的日志目录或日志文件，实时跟踪并读取这些文件的更新内容，并将这些数据发送到指定的输出目标，例如Elasticsearch或Kafka等。

这里选择Filebeat进行日志采集的主要原因在于其资源消耗极低，相较于Flume和Logstash，Filebeat占用的内存最少，对CPU的负载也最小。它的运行进程十分稳定，很少出现崩溃或宕机的情况。

首先下载Filebeat

curl -L-O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.12.0-linux-x86_64.tar.gz

解压缩文件

tar xzvf filebeat-8.12.0-linux-x86_64.tar.gz

进入目录

cd filebeat-8.12.0-linux-x86_64

编写配置文件接入Kafka

vim filebeat.yaml

filebeat.yaml的文件内容

filebeat.inputs:-type: log
  paths:- /doc/input/*.log# 更换为你的日志文件路径processors:-include_fields:fields:["message"]output.kafka:# 更换为你的Kafka地址和主题.hosts:["192.168.235.130:9092"]topic: k2gg
  codec:format:string:'%{[message]}'

运行Filebeat采集日志

./filebeat -e -c ./filebeat.yaml

这是log日志的信息，现要求保持原始格式发送到Kafka
Filebeat采集日志信息发送到Kafka的主题，消费者收到的信息如下，Filebeat会添加一些自带的数据，比如时间戳和元数据等，但是一般情况下只需要采集message里面的信息，通过filebeat.yaml中的processors和codec即可实现。
processors处理只保留 message 的字段信息，其他字段将被丢弃，codec用于定义数据的编码格式，将 message 字段的值作为字符串发送到 kafka，这样就可以

保留日志信息的原始格式发送到Kafka。

消费者消费原始格式的日志消息

Flink消费Kafka实时写入Doris

在写入之前，建立doris的数据表用于接收消费的信息

CREATE TABLE transactions (
  timestamp datetime,
  user_id INT,
  transaction_type VARCHAR(50),
  amount DECIMAL(15,2),
  currency CHAR(3),
  status VARCHAR(20),
  description TEXT
)
DISTRIBUTED BYHASH(user_id) BUCKETS 10PROPERTIES("replication_num"="1");

引入依赖

<dependency><groupId>org.apache.doris</groupId><artifactId>flink-doris-connector-1.16</artifactId><version>24.0.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-core</artifactId><version>1.17.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>1.17.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.17.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>1.17.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-common</artifactId><version>1.17.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java</artifactId><version>1.17.0</version></dependency>

主程序

packageflink;importorg.apache.doris.flink.cfg.DorisExecutionOptions;importorg.apache.doris.flink.cfg.DorisOptions;importorg.apache.doris.flink.cfg.DorisReadOptions;importorg.apache.doris.flink.sink.DorisSink;importorg.apache.doris.flink.sink.writer.serializer.SimpleStringSerializer;importorg.apache.flink.api.common.functions.MapFunction;importorg.apache.flink.api.common.serialization.SimpleStringSchema;importorg.apache.flink.streaming.api.datastream.DataStream;importorg.apache.flink.streaming.api.datastream.DataStreamSource;importorg.apache.flink.streaming.api.environment.CheckpointConfig;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;importorg.apache.flink.table.api.DataTypes;importorg.apache.flink.table.types.DataType;importjava.util.Properties;publicclassDorisWrite{publicstaticvoidmain(String[] args)throwsException{Properties props =newProperties();//Kafka broker的地址
       props.put("bootstrap.servers","192.168.235.130:9092");
       props.put("group.id","test");

        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset","latest");StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//指定消费的主题FlinkKafkaConsumer<String> flinkKafkaConsumer =newFlinkKafkaConsumer<>("k2gg",newSimpleStringSchema(),props);DorisSink.Builder<String> builder =DorisSink.builder();DorisOptions.Builder dorisBuilder =DorisOptions.builder();//Doris的地址以及账号密码等信息
        dorisBuilder.setFenodes("192.168.235.130:8030").setTableIdentifier("test.transactions").setUsername("root").setPassword("1445413748");Properties pro =newProperties();
        pro.setProperty("format","json");
        pro.setProperty("read_json_by_line","true");DorisExecutionOptions  executionOptions =DorisExecutionOptions.builder().setLabelPrefix("label-doris12"+System.currentTimeMillis())//streamload label prefix,.setStreamLoadProp(pro).build();

        builder.setDorisReadOptions(DorisReadOptions.builder().build()).setDorisExecutionOptions(executionOptions).setSerializer(newSimpleStringSerializer())//serialize according to string.setDorisOptions(dorisBuilder.build());DataStreamSource<String> dataStreamSource = env.addSource(flinkKafkaConsumer);// 将Kafka数据转换为JSON格式DataStream<String> jsonStream = dataStreamSource.map(newMapFunction<String,String>(){@OverridepublicStringmap(String value)throwsException{System.out.println("value"+value);// 分割字符串String[] parts = value.split(",");// 创建JSON字符串StringBuilder jsonString =newStringBuilder();
                jsonString.append("{");
                jsonString.append("\"timestamp\":\"").append(parts[0]).append("\",");
                jsonString.append("\"user_id\":").append(parts[1]).append(",");
                jsonString.append("\"transaction_type\":\"").append(parts[2]).append("\",");
                jsonString.append("\"amount\":").append(parts[3]).append(",");
                jsonString.append("\"currency\":\"").append(parts[4]).append("\",");
                jsonString.append("\"status\":\"").append(parts[5]).append("\",");
                jsonString.append("\"description\":\"").append(parts[6].replace("\"","")).append("\"");
                jsonString.append("}");return jsonString.toString();}});

      
        jsonStream.print();
        jsonStream.sinkTo(builder.build());

        env.execute("flink kafka to doris by datastream");}}

运行主程序通过Flink消费Kafka的信息写入doris
log日志的信息

登录Doris进行验证

mysql -h k8s-master -P9030-uroot -p

这是没运行主程序之前doris的数据，没有2024-10-15这一天的数据。

select * from transactions where date(timestamp)="2024-10-15";

运行主程序之后，Flink将Kafka主题的信息实时写入Doris。

总结

1.Filebeat格式问题
Filebeat采集日志格式会添加一些自带的额外信息，一般情况下只需要message里面的字段信息，那么yaml文件配置processors和codec属性即可。

processors

处理只保留

message

的字段信息，其他字段将被丢弃，codec用于定义数据的编码格式，将 message 字段的值作为字符串发送到 kafka，这样就可以保留日志信息的原始格式发送到Kafka。
2.Flink消费Kafka失败
Flink在消费Kafka主题的过程中，不要往该主题发送其他格式的数据，否则会解析失败，尽量新建一个新主题来接收Filebeat采集过来的日志信息。如果还是执行失败，可以尝试在

setLabelPrefix

添加一个时间戳，这样保证每次生成的标签前缀都不一样，这是因为客户端会生成一个唯一的标签来标识这次导入Doris的操作，Doris服务器会根据这个标签来跟踪导入的进度和状态，如果导入过程中出现问题，Doris会保留失败的数据，客户端就可以通过标签重新导入这些数据。
3.实时写入Doris失败
Flink处理字段的数据类型要与Doris匹配，可以参考官方文档Doris 和 Flink 列类型映射关系。