kafka sink

Author: zhangblue

src/main/resources/kafka/home-producer.properties

@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# see kafka.producer.ProducerConfig for more details
+
+############################# Producer Basics #############################
+
+# list of brokers used for bootstrapping knowledge about the rest of the cluster
+# format: host1:port1,host2:port2 ...
+bootstrap.servers=192.168.247.103:9092
+
+# specify the compression codec for all data generated: none, gzip, snappy, lz4
+compression.type=none
+
+# name of the partitioner class for partitioning events; default partition spreads data randomly
+#partitioner.class=
+
+# the maximum amount of time the client will wait for the response of a request
+#request.timeout.ms=
+
+# how long `KafkaProducer.send` and `KafkaProducer.partitionsFor` will block for
+#max.block.ms=
+
+# the producer will wait for up to the given delay to allow other records to be sent so that the sends can be batched together
+#linger.ms=
+
+# the maximum size of a request in bytes
+#max.request.size=
+
+# the default batch size in bytes when batching multiple records sent to a partition
+#batch.size=
+
+# the total bytes of memory the producer can use to buffer records waiting to be sent to the server
+#buffer.memory=

src/main/scala/com/zhangblue/sink/SinkKafkaDemo.scala

@@ -2,11 +2,9 @@ package com.zhangblue.sink
 
 import java.util.Properties
 
-import com.alibaba.fastjson.JSONObject
 import com.zhangblue.entity.TemperatureSensor
 import org.apache.flink.api.common.serialization.{DeserializationSchema, SimpleStringSchema}
 import org.apache.flink.streaming.api.scala._
-import org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaProducer
 import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}
 
 /**
@@ -21,11 +19,14 @@ object SinkKafkaDemo {
     //需求，求出传感器迄今为止探索到的最高温度的传感器数据
     //1. 环境
     val env = StreamExecutionEnvironment.getExecutionEnvironment
+    //获取类加载器
+    val classLoader = this.getClass.getClassLoader;
+
     //2. 读取kafka源，根据传感器的id进行分组，求每组中最大的温度值，并显示结果
     val topic = "my-topic"
     val valueDeserializationSchema: DeserializationSchema[String] = new SimpleStringSchema()
     val props: Properties = new Properties()
-    props.load(this.getClass.getClassLoader.getResourceAsStream("kafka/localhost-consumer.properties"))
+    props.load(classLoader.getResourceAsStream("kafka/home-consumer.properties"))
     val kafkaSource = env.addSource(new FlinkKafkaConsumer011[String](topic, valueDeserializationSchema, props))
 
     val srcDataStream: DataStream[TemperatureSensor] = kafkaSource.filter(_.trim.nonEmpty).map(fun = lineData => {
@@ -39,21 +40,12 @@ object SinkKafkaDemo {
     }).keyBy("id").maxBy("temperature")
 
 
-    val producerConfig:Properties = new Properties();
-    producerConfig.load(this.getClass.getClassLoader.getResourceAsStream("kafka/localhost-producer.properties"))
+    //将DataStream 输出到Kafka中
+    val producerConfig: Properties = new Properties();
+    producerConfig.load(classLoader.getResourceAsStream("kafka/home-producer.properties"))
     srcDataStream.map(data => data.toString).addSink {
-      sinkFunction = new FlinkKafkaProducer[String]("target-topic", new SimpleStringSchema(), new Properties())
+      new FlinkKafkaProducer011[String]("flink-target-topic", new SimpleStringSchema(), producerConfig)
     }
-
-
-
-
-
-
-
-
-
-
     //3. 启动
     env.execute(this.getClass.getSimpleName)
   }