Flink 输出至 Elasticsearch

这篇具有很好参考价值的文章主要介绍了Flink 输出至 Elasticsearch。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

【1】引入pom.xml依赖

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

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

package com.zzx.flink

import java.util

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.client.Requests


object EsSinkTest {
  def main(args: Array[String]): Unit = {
    // 创建一个流处理执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //从文件中读取数据并转换为 类
    val inputStreamFromFile: DataStream[String] = env.readTextFile("E:\\Project\\flink\\src\\main\\resources\\wordcount.txt")
    //转换
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map( data => {
        var dataArray = data.split(",")
        SensorReading(dataArray(0),dataArray(1).toLong,dataArray(2).toDouble)
      })

    //定义一个 HttpHosts
    val httpHost = new util.ArrayList[HttpHost]()
    //默认 9200 我的修改为了 9201
    httpHost.add(new HttpHost("192.168.1.12",9200,"http"))
    httpHost.add(new HttpHost("127.0.0.1",9200,"http"))
    //定义一个 ElasticSearchFuntion 操作 es的function
    val esSinkFunc = new ElasticsearchSinkFunction[SensorReading] {
      //element 每一条数据 通过 index 发送
      override def process(element: SensorReading, runtimeContext: RuntimeContext, index: RequestIndexer): Unit = {
        //包装写入 es 的数据
        val dataSource = new util.HashMap[String,String]()
        dataSource.put("sensor_id",element.id)
        dataSource.put("temp",element.temperature.toString)
        dataSource.put("ts",element.timestamp.toString)

        //index
        val indexRequest = Requests.indexRequest()
            .index("sensor_temp")
            .`type`("readingdata")
            .source(dataSource)
        index.add(indexRequest)
        println("saved successfully " + element.toString)
      }
    }
    //输出值 es
    dataStream.addSink(new ElasticsearchSink.Builder[SensorReading](httpHost,esSinkFunc).build())
    env.execute("es")
  }
}

【3】ES6输出展示

Flink 输出至 Elasticsearch,Flink,flink,elasticsearch,大数据,java,面试,后端,性能优化文章来源地址https://www.toymoban.com/news/detail-766525.html

到了这里,关于Flink 输出至 Elasticsearch的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Flink 系例 之 Connectors 连接 ElasticSearch

    通过使用 Flink DataStream Connectors 数据流连接器连接到 ElasticSearch 搜索引擎的文档数据库 Index,并提供数据流输入与输出操作; 示例环境 示例数据源 (项目码云下载) Flink 系例 之 搭建开发环境与数据 示例模块 (pom.xml) Flink 系例 之 DataStream Connectors 与 示例模块 数据流输入

    2024年02月16日
    浏览(47)
  • SpringData、SparkStreaming和Flink集成Elasticsearch

    本文代码链接: https://download.csdn.net/download/shangjg03/88522188 Spring Data是一个用于简化数据库、非关系型数据库、索引库访问,并支持云服务的开源框架。其主要目标是使得对数据的访问变得方便快捷,并支持map-reduce框架和云计算数据服务。 Spring Data可以极大的简化JPA(Elast

    2024年02月05日
    浏览(35)
  • flink日志实时采集写入Kafka/ElasticSearch

    由于公司想要基于flink的日志做实时预警功能,故需要实时接入,并刷入es进行分析。 日志接入必须异步,不能影响服务性能 kafka集群宕机,依旧能够提交flink任务且运行任务 kafka集群挂起恢复,可以依旧续写实时运行日志 在类上加上@Plugin注解,标记为自定义appender 在类加上

    2024年02月08日
    浏览(53)
  • Flink系列之:Elasticsearch SQL 连接器

    Sink: Batch Sink: Streaming Append Upsert Mode Elasticsearch 连接器允许将数据写入到 Elasticsearch 引擎的索引中。本文档描述运行 SQL 查询时如何设置 Elasticsearch 连接器。 连接器可以工作在 upsert 模式,使用 DDL 中定义的主键与外部系统交换 UPDATE/DELETE 消息。 如果 DDL 中没有定义主键,那么

    2024年02月04日
    浏览(54)
  • Flink CDC 系列(7)—— 从 MySQL 到 ElasticSearch

    Flink CDC 系列文章: 《Flink CDC 系列(1)—— 什么是 Flink CDC》 《Flink CDC 系列(2)—— Flink CDC 源码编译》 《Flink CDC 系列(3)—— Flink CDC MySQL Connector 与 Flink SQL 的结合使用案例Demo》 《Flink CDC 系列(4)—— Flink CDC MySQL Connector 常用参数表》 《Flink CDC 系列(5)—— Flink

    2023年04月09日
    浏览(28)
  • flink-cdc同步mysql数据到elasticsearch

    CDC是(Change Data Capture 变更数据获取)的简称。核心思想是,监测并捕获数据库的变动(包括数据 或 数据表的插入INSERT、更新UPDATE、删除DELETE等),将这些变更按发生的顺序完整记录下来,写入到消息中间件中以供其他服务进行订阅及消费。 cdc项目地址:https://github.com/ver

    2024年02月13日
    浏览(81)
  • Flink CDC系列之:TiDB CDC 导入 Elasticsearch

    替换目录 tidb-docker-compose 里面的 docker-compose.yml 文件,内容如下所示: 该 Docker Compose 中包含的容器有: TiDB 集群: tikv、pd、tidb。 Elasticsearch:orders 表将和 products 表进行 join,join 的结果写入 Elasticsearch 中。 Kibana:可视化 Elasticsearch 中的数据。 本机添加 host 映射 pd 和 tikv 映射

    2024年02月12日
    浏览(28)
  • Flink CDC系列之:Oracle CDC 导入 Elasticsearch

    Flink CDC系列之:Oracle CDC Connector 该 Docker Compose 中包含的容器有: Oracle: Oracle 11g, 已经预先创建了 products 和 orders表,并插入了一些数据 Elasticsearch: orders 表将和 products 表进行join,join的结果写入Elasticsearch中 Kibana: 可视化 Elasticsearch 中的数据 在 docker-compose.yml 所在目录下运行如下

    2024年02月12日
    浏览(32)
  • 最新版Flink CDC MySQL同步Elasticsearch(一)

    首先我们要基于Flink CDC MySQL同步MySQL的环境基础上(flink-1.17.1、Java8、MySQL8)搭建Elasticsearch7-17-10和Kibana 7.17.10。笔者已经搭建好环境,这里不做具体演示了,如果需要Es的搭建教程情况笔者其他博客 注意: 建议生产环境统一使用稳定版本Flink1.16.*。笔者这里只是作为教程编写

    2024年02月13日
    浏览(35)
  • 实战:大数据Flink CDC同步Mysql数据到ElasticSearch

    前面的博文我们分享了大数据分布式流处理计算框架Flink和其基础环境的搭建,相信各位看官都已经搭建好了自己的运行环境。那么,今天就来实战一把使用Flink CDC同步Mysql数据导Elasticsearch。 CDC简介 CDC 的全称是 Change Data Capture(变更数据捕获技术) ,在广义的概念上,只要

    2024年02月09日
    浏览(43)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包