本文模拟实际生产环境，通过FileBeat采集日志信息到Kafka，再通过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 BY HASH(user_id) BUCKETS 10
PROPERTIES("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>
主程序
package flink;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.serializer.SimpleStringSerializer;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.types.DataType;
import java.util.Properties;
public class DorisWrite {
public static void main(String[] args) throws Exception {
Properties props = new Properties();
//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 = new FlinkKafkaConsumer<>("k2gg",new SimpleStringSchema(),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 = new Properties();
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(new SimpleStringSerializer()) //serialize according to string
.setDorisOptions(dorisBuilder.build());
DataStreamSource<String> dataStreamSource = env.addSource(flinkKafkaConsumer);
// 将Kafka数据转换为JSON格式
DataStream<String> jsonStream = dataStreamSource.map(new MapFunction<String, String>() {
@Override
public String map(String value) throws Exception {
System.out.println("value"+value);
// 分割字符串
String[] parts = value.split(",");
// 创建JSON字符串
StringBuilder jsonString = new StringBuilder();
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 -P 9030 -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 列类型映射关系。
