kafka flumn sparkstreaming java实现监听文件夹内容保存到Phoenix中

#指定agent的sources（如r1、r2）、sinks（如k1、k2）、channels（如c1、c2）

tier1.sources=r1 r2

tier1.sinks=k1 k2

tier1.channels=c1 c2

#描述source r1

#配置目录scource

#配置监控的目录，当目录出现新文件时会进行写入

tier1.sources.r1.type = spooldir

tier1.sources.r1.spoolDir =/var/flumefile

tier1.sources.r1.channels=c1

#配置数据源输出

#设置Kafka接收器,此处最坑,注意版本,此处为Flume 1.6.0的输出槽类型

tier1.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink

#设置Kafka的broker地址和端口号

tier1.sinks.k1.brokerList=master:9092

#设置Kafka的Topic

tier1.sinks.k1.topic=topic-test

#设置序列化方式

tier1.sinks.k1.serializer.class=kafka.serializer.StringEncoder

tier1.sinks.k1.channel=c1

#配置channels类型为 File

tier1.channels.c1.type=memory

tier1.channels.c1.capacity=10000

tier1.channels.c1.transactionCapacity=100

#描述source r1

#配置目录scource

#配置监控的目录，当目录出现新文件时会进行写入

tier1.sources.r2.type = spooldir

tier1.sources.r2.spoolDir =/var/flumefile2

tier1.sources.r2.channels=c2

#配置数据源输出

#设置Kafka接收器,此处最坑,注意版本,此处为Flume 1.6.0的输出槽类型

tier1.sinks.k2.type= org.apache.flume.sink.kafka.KafkaSink

#设置Kafka的broker地址和端口号

tier1.sinks.k2.brokerList=master:9092

#设置Kafka的Topic

tier1.sinks.k2.topic=test

#设置序列化方式

tier1.sinks.k2.serializer.class=kafka.serializer.StringEncoder

tier1.sinks.k2.channel=c2

#配置channels类型为 File

tier1.channels.c2.type=memory

tier1.channels.c2.capacity=10000

tier1.channels.c2.transactionCapacity=100

<groupId>org.apache.spark</groupId>

<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>

<version>2.3.1</version>

</dependency>

</dependencies>

String brokers = "master:9092";

SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("streamingKafka");

JavaSparkContext sc = new JavaSparkContext(conf);

sc.setLogLevel("WARN");

JavaStreamingContext ssc = new JavaStreamingContext(sc, Durations.seconds(10));

// kafka相关参数，必要！缺了会报错

Map<String, Object> kafkaParams = new HashMap<>();

kafkaParams.put("bootstrap.servers", brokers);

kafkaParams.put("key.deserializer", StringDeserializer.class);

kafkaParams.put("value.deserializer", StringDeserializer.class);

kafkaParams.put("group.id", "cloudera_mirrormaker");

// Topic分区 也可以通过配置项实现

// 如果没有初始化偏移量或者当前的偏移量不存在任何服务器上，可以使用这个配置属性

// earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费

// latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据

// none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

kafkaParams.put("auto.offset.reset", "latest");

// 如果是使用spark-streaming-kafka-0-10，那么我们建议将enable.auto.commit设为false。

// 这个配置只是在这个版本生效，enable.auto.commit如果设为true的话，那么意味着offsets会按照auto.commit.interval.ms中所配置的间隔来周期性自动提交到Kafka中。

// 在Spark Streaming中，将这个选项设置为true的话会使得Spark应用从kafka中读取数据之后就自动提交，

// 而不是数据处理之后提交，这不是我们想要的。所以为了更好地控制offsets的提交，我们建议将enable.auto.commit设为false。

kafkaParams.put("enable.auto.commit", false);

// 设置消费的topic

Collection<String> topicsColl = Arrays.asList("topic-test", "test");

// 获取DStream

JavaInputDStream<ConsumerRecord<String, String>> lines = KafkaUtils.createDirectStream(ssc,

LocationStrategies.PreferConsistent(), ConsumerStrategies.Subscribe(topicsColl, kafkaParams));

// 遍历JavaInputDStream的到RDD

lines.foreachRDD(rdd -> {

// 遍历RDD得到每一列的信息

rdd.foreach(x -> {

// 获取行

String value = x.value();

// 获取需要保存的每一行的信息

String[] split = value.split(";");

// 判断是否为保存数据需要的格式

// 保存offset，保证每次获取新数据

//获取偏移量

OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();

 //保存偏移量到Kafka

((CanCommitOffsets) lines.inputDStream()).commitAsync(offsetRanges);

});

ssc.start();

try {

ssc.awaitTermination();

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

topic = test,

partition = 0,

offset = 547,

CreateTime = 1558419603357,

checksum = 890333719,

serialized key size = -1,

serialized value size = 92,

key = null,

value = 对应该行的数据