[大数据] Flink elasticSearchSink

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Flink elasticSearchSink -> 正文阅读

[大数据]Flink elasticSearchSink

<dependency>
? ? <groupId>org.apache.flink</groupId>
? ? <artifactId>flink-connector-elasticsearch6_2.12</artifactId>
? ? <version>1.14.4</version>
</dependency>

SinkFunction实现类：

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}

class ElasticIndexSinkFunction(indexType:String) extends ElasticsearchSinkFunction[JSONObject]{
  override def process(element: JSONObject, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
    try{        
            val uniqueKey=element.getString("uniqueKey")
            val indexName=EsIndexName.getWeekMondayIndexName(indexType,element.getLongValue("startTime"))
            val request: IndexRequest = Requests.indexRequest
            .index(indexName)
            .`type`("_doc")
            .id(uniqueKey)
            .source(element)

      requestIndexer.add(request)
    }catch {
      case e:Exception=>e.printStackTrace()
    }

  }
}

数据写入es:

import org.apache.commons.configuration2.builder.fluent.Configurations
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost

import java.util
import java.util.UUID

object EsSlinkUtil {
  def sendData(indexType: String, data: DataStream[JSONObject], jobType: String) = {
    val configs = new Configurations()
    val configuration = configs.properties("SysConfig.properties")
    val esIP = configuration.getString("eSPort_export")
    val httpHosts = new util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost(esIP, 9200, "http"))
    try {
      val elasticSink: ElasticsearchSink.Builder[JSONObject] = new ElasticsearchSink.Builder[JSONObject](httpHosts, new ElasticIndexSinkFunction(indexType))
      data.addSink(elasticSink.build())
        .name(indexType + "-" + UUID.randomUUID().toString.replaceAll("-", ""))
        .uid(indexType + "-" + jobType)
    } catch {
      case e: Exception => e.printStackTrace()
    }
  }
}