Flink 输出至 Elasticsearch

【1】引入pom.xml依赖

1. <dependency>
2.     <groupId>org.apache.flink</groupId>
3.     <artifactId>flink-connector-elasticsearch6_2.12</artifactId>
4.     <version>1.10.0</version>
5. </dependency>

复制代码

【2】ES6 Scala代码，主动导入的scala包需要修改为scala._ 否则会出现错误。

1. package com.zzx.flink
3. import java.util
5. import org.apache.flink.api.common.functions.RuntimeContext
6. import org.apache.flink.streaming.api.scala._
7. import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
8. import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
9. import org.apache.http.HttpHost
10. import org.elasticsearch.client.Requests
13. object EsSinkTest {
14.   def main(args: Array[String]): Unit = {
15.     // 创建一个流处理执行环境
16.     val env = StreamExecutionEnvironment.getExecutionEnvironment
17.     //从文件中读取数据并转换为 类
18.     val inputStreamFromFile: DataStream[String] = env.readTextFile("E:\\Project\\flink\\src\\main\\resources\\wordcount.txt")
19.     //转换
20.     val dataStream: DataStream[SensorReading] = inputStreamFromFile
21.       .map( data => {
22.         var dataArray = data.split(",")
23.         SensorReading(dataArray(0),dataArray(1).toLong,dataArray(2).toDouble)
24.       })
26.     //定义一个 HttpHosts
27.     val httpHost = new util.ArrayList[HttpHost]()
28.     //默认 9200 我的修改为了 9201
29.     httpHost.add(new HttpHost("192.168.1.12",9200,"http"))
30.     httpHost.add(new HttpHost("127.0.0.1",9200,"http"))
31.     //定义一个 ElasticSearchFuntion 操作 es的function
32.     val esSinkFunc = new ElasticsearchSinkFunction[SensorReading] {
33.       //element 每一条数据 通过 index 发送
34.       override def process(element: SensorReading, runtimeContext: RuntimeContext, index: RequestIndexer): Unit = {
35.         //包装写入 es 的数据
36.         val dataSource = new util.HashMap[String,String]()
37.         dataSource.put("sensor_id",element.id)
38.         dataSource.put("temp",element.temperature.toString)
39.         dataSource.put("ts",element.timestamp.toString)
41.         //index
42.         val indexRequest = Requests.indexRequest()
43.             .index("sensor_temp")
44.             .`type`("readingdata")
45.             .source(dataSource)
46.         index.add(indexRequest)
47.         println("saved successfully " + element.toString)
48.       }
49.     }
50.     //输出值 es
51.     dataStream.addSink(new ElasticsearchSink.Builder[SensorReading](httpHost,esSinkFunc).build())
52.     env.execute("es")
53.   }
54. }

复制代码

【3】ES6输出展示

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。