1、 在主节点使用Flume采集/data_log目录下实时日志文件中的数据，将数据存入到Kafka的Topic中（Topic名称分别为ChangeRecord、ProduceRecord和EnvironmentData，分区数为4）

好的，这是一个关于Flume、Kafka的数据采集和存储问题，你需要在主节点上配置Flume并使用它来采集/data_log目录下的实时日志文件，并将数据存储到Kafka的三个Topic中。具体的步骤如下：

1. 在主节点上安装Flume和Kafka，并确保它们能够正常工作。

2. 在Flume的配置文件中定义一个source，用于采集/data_log目录下的实时日志文件。例如：

# Define a source for tailing log files
agent.sources = logsource
agent.sources.logsource.type = exec
agent.sources.logsource.command = tail -F /data_log/*.log

3. 在Flume的配置文件中定义一个channel，用于存储采集到的数据。例如：

# Define a channel for storing events
agent.channels = logchannel
agent.channels.logchannel.type = memory
agent.channels.logchannel.capacity = 10000

4. 在Flume的配置文件中定义三个sink，分别用于将采集到的数据存储到Kafka的三个Topic中。例如：

# Define sinks for sending events to Kafka
agent.sinks.kafka1.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka1.topic = ChangeRecord
agent.sinks.kafka1.brokerList = kafka1:9092,kafka2:9092,kafka3:9092
agent.sinks.kafka1.batchSize = 100

agent.sinks.kafka2.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka2.topic = ProduceRecord
agent.sinks.kafka2.brokerList = kafka1:9092,kafka2:9092,kafka3:9092
agent.sinks.kafka2.batchSize = 100

agent.sinks.kafka3.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka3.topic = EnvironmentData
agent.sinks.kafka3.brokerList = kafka1:9092,kafka2:9092,kafka3:9092
agent.sinks.kafka3.batchSize = 100

其中，brokerList指定了Kafka的地址和端口号，batchSize表示每次向Kafka发送的事件的数量。

5. 在Flume的配置文件中定义一个agent，将source、channel和sink组合起来。例如：

# Define the agent and the flow of events
agent.sources = logsource
agent.channels = logchannel
agent.sinks = kafka1 kafka2 kafka3

agent.sources.logsource.channels = logchannel
agent.sinks.kafka1.channel = logchannel
agent.sinks.kafka2.channel = logchannel
agent.sinks.kafka3.channel = logchannel

6. 启动Flume，并检查是否能够采集到/data_log目录下的实时日志文件，并将数据存储到Kafka的三个Topic中。

以上就是将Flume和Kafka用于数据采集和存储的一个简单示例。具体的配置可能会因为环境和需求的不同而有所不同，需要根据实际情况进行调整。