Kafka集群间同步数据方案-Flume_kafka 数据同步

# 新建topic
./bin/kafka-topics.sh --create --bootstrap-server 10.19.162.107:9092 --replication-factor 2 --partitions 1 --topic test-flume-logs1

import os

from pykafka import KafkaClient

dirPath = 'E:\\log\\测点数据'
fileList = os.listdir(dirPath)

client = KafkaClient(hosts="10.19.162.107:9092,10.19.162.108:9092,10.19.162.109:9092")

topic = client.topics['test-flume-logs1']

print(len(fileList))

f2 = open('E:\\log\\all.txt',mode="w",encoding='utf-8')


for j in range(10):
    with topic.get_sync_producer() as producer:
        for i in range(len(fileList)):
            fileName = dirPath + "\\" + fileList[i]
            f = open(fileName,encoding='utf-8')

            line = f.readline()
            # 以下是文件读取伪代码
            while line:
                line = line + '\n'
                f2.write(line)
                # 写入kafka
                producer.produce(line.encode())
                # print(line)
                line = f.readline()
            f.close()

f2.close()

#kafka相关source和sink参数可以参考如下网址
# 中文翻译版：https://flume.liyifeng.org/#kafka-sink
# 英文原版： https://flume.apache.org/releases/content/1.10.1/FlumeUserGuide.html#kafka-sink
# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1


# Describe/configure the source
a2.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a2.sources.r1.batchSize = 1000
a2.sources.r1.batchDurationMillis = 2000
# 此处hadoop01,hadoop02,hadoop03在host文件中已配置，ip分别为10.19.162.107,10.19.162.108,10.19.162.109
a2.sources.r1.kafka.bootstrap.servers = hadoop01:9092,hadoop02:9092,hadoop03:9092
a2.sources.r1.kafka.topics = test-flume-logs1
a2.sources.r1.kafka.consumer.group.id = testgruoup
a2.sources.r1.migrateZookeeperOffsets = false
a2.sources.r1.kafka.consumer.auto.offset.reset = earliest

## Source 拦截器，如果不指定拦截器，flume有bug，sink会使用最上方配置的kafka配置
a2.sources.r1.interceptors = i1
a2.sources.r1.interceptors.i1.type = static
a2.sources.r1.interceptors.i1.key = topic
a2.sources.r1.interceptors.i1.preserveExisting = false
a2.sources.r1.interceptors.i1.value = test-flume-logs1

# Describe the sink
a2.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a2.sinks.k1.kafka.topic = test-flume-logs1
a2.sinks.k1.kafka.bootstrap.servers = 10.19.162.101:9092,10.19.162.102:9092,10.19.162.105:9092
a2.sinks.k1.kafka.flumeBatchSize = 20
a2.sinks.k1.kafka.producer.acks = 1
a2.sinks.k1.kafka.producer.linger.ms = 1
a2.sinks.ki.kafka.producer.compression.type = snappy

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

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

 bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 10.19.162.107:9092 --topic test-flume-logs1  --time -1

# 其中9001位flume的状态监控端口
bin/flume-ng agent --conf conf --conf-file conf/kafka-to-kafka.conf --name a2 -Dflume.root.logger=INFO,console -Dflume.monitoring.type=http -Dflume.monitoring.port=9001

{
	"SINK.k1": {
		"ConnectionCreatedCount": "0",
		"BatchCompleteCount": "0",
		"BatchEmptyCount": "26",
		"EventDrainAttemptCount": "100000", # event数量，也就是从源kafka的topic中读取的数量
		"StartTime": "1666580989920",
		"BatchUnderflowCount": "0",
		"ConnectionFailedCount": "0",
		"ConnectionClosedCount": "0",
		"Type": "SINK",
		"RollbackCount": "0",
		"EventDrainSuccessCount": "100000",	# 成功写入的数量
		"KafkaEventSendTimer": "9861",
		"StopTime": "0"
	},
	"CHANNEL.c1": {
		"ChannelCapacity": "2000",
		"ChannelFillPercentage": "0.0",
		"Type": "CHANNEL",
		"ChannelSize": "0",
		"EventTakeSuccessCount": "100000",
		"EventTakeAttemptCount": "100027",
		"StartTime": "1666580989317",
		"EventPutAttemptCount": "100000",
		"EventPutSuccessCount": "100000",
		"StopTime": "0"
	},
	"SOURCE.r1": {
		"KafkaEventGetTimer": "4470",
		"AppendBatchAcceptedCount": "0",
		"EventAcceptedCount": "100000",
		"AppendReceivedCount": "0",
		"StartTime": "1666580989944",
		"AppendBatchReceivedCount": "0",
		"KafkaCommitTimer": "257",
		"EventReceivedCount": "100000",
		"Type": "SOURCE",
		"AppendAcceptedCount": "0",
		"OpenConnectionCount": "0",
		"KafkaEmptyCount": "30",
		"StopTime": "0"
	}
}