当前位置: 首页 > news >正文

flink如何写入es

提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档

文章目录

  • 前言
  • 一、写入到Elasticsearch5
  • 二、写入到Elasticsearch7
  • 总结


前言

Flink sink 流数据写入到es5和es7的简单示例。


一、写入到Elasticsearch5

  • pom maven依赖
        <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-elasticsearch5_2.11</artifactId><version>${flink.version}</version></dependency>
  • 代码如下(示例):
public class Es5SinkDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();Row row=Row.of("张三","001",getTimestamp("2016-10-24 21:59:06"));Row row2=Row.of("张三","002",getTimestamp("2016-10-24 21:50:06"));Row row3=Row.of("张三","002",getTimestamp("2016-10-24 21:51:06"));Row row4=Row.of("李四","003",getTimestamp("2016-10-24 21:50:56"));Row row5=Row.of("李四","004",getTimestamp("2016-10-24 00:48:36"));Row row6=Row.of("王五","005",getTimestamp("2016-10-24 00:48:36"));DataStreamSource<Row> source =env.fromElements(row,row2,row3,row4,row5,row6);Map<String, String> config = new HashMap<>();
//        config.put("cluster.name", "my-cluster-name");
//        config.put("bulk.flush.max.actions", "1");List<InetSocketAddress> transportAddresses = new ArrayList<>();transportAddresses.add(new InetSocketAddress(InetAddress.getByName("10.68.8.60"), 9300));//Sink操作DataStreamSink<Row> rowDataStreamSink = source.addSink(new ElasticsearchSink<>(config, transportAddresses, new ElasticsearchSinkFunction<Row>() {public IndexRequest createIndexRequest(Row element) {Map<String, Object> json = new HashMap<>();json.put("name22", element.getField(0).toString());json.put("no22", element.getField(1));json.put("age", 34);json.put("create_time", element.getField(2));return Requests.indexRequest().index("cc").type("mtype").id(element.getField(1).toString()).source(json);}@Overridepublic void process(Row element, RuntimeContext ctx, RequestIndexer indexer) {//利用requestIndexer进行发送请求,写入数据indexer.add(createIndexRequest(element));}}));env.execute("es demo");}private static java.sql.Timestamp getTimestamp(String str) throws Exception {
//		String string = "2016-10-24 21:59:06";SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");java.util.Date date=sdf.parse(str);java.sql.Timestamp s = new java.sql.Timestamp(date.getTime());return s;}

二、写入到Elasticsearch7

  • pom maven依赖
        <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-elasticsearch7_2.11</artifactId><version>${flink.version}</version><scope>provided</scope></dependency>
  • 代码如下(示例):
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.types.Row;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class EsSinkDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();Row row=Row.of("张三","001",getTimestamp("2016-10-24 21:59:06"));Row row2=Row.of("张三","002",getTimestamp("2016-10-24 21:50:06"));Row row3=Row.of("张三","002",getTimestamp("2016-10-24 21:51:06"));Row row4=Row.of("李四","003",getTimestamp("2016-10-24 21:50:56"));Row row5=Row.of("李四","004",getTimestamp("2016-10-24 00:48:36"));Row row6=Row.of("王五","005",getTimestamp("2016-10-24 00:48:36"));DataStreamSource<Row> source =env.fromElements(row,row2,row3,row4,row5,row6);Map<String, String> config = new HashMap<>();
//        config.put("cluster.name", "my-cluster-name");
// This instructs the sink to emit after every element, otherwise they would be buffered
//        config.put("bulk.flush.max.actions", "1");List<HttpHost> hosts = new ArrayList<>();hosts.add(new HttpHost("10.68.8.69",9200,"http"));ElasticsearchSink.Builder<Row> esSinkBuilder = new ElasticsearchSink.Builder<Row>(hosts,new ElasticsearchSinkFunction<Row>() {public IndexRequest createIndexRequest(Row element) {Map<String, Object> json = new HashMap<>();json.put("name22", element.getField(0).toString());json.put("no22", element.getField(1));json.put("age", 34);
//                json.put("create_time", element.getField(2));return Requests.indexRequest().index("cc").id(element.getField(1).toString()).source(json);}@Overridepublic void process(Row element, RuntimeContext ctx, RequestIndexer indexer) {//利用requestIndexer进行发送请求,写入数据indexer.add(createIndexRequest(element));}});esSinkBuilder.setBulkFlushMaxActions(100);//Sink操作source.addSink(esSinkBuilder.build());env.execute("es demo");}private static java.sql.Timestamp getTimestamp(String str) throws Exception {
//		String string = "2016-10-24 21:59:06";SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");java.util.Date date=sdf.parse(str);java.sql.Timestamp s = new java.sql.Timestamp(date.getTime());return s;}
}

总结

flink写入es5和es7 的区别是引入不同的flink-connector-elasticsearch,es7已没有type的概念故无需再设置type。

相关文章:

flink如何写入es

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、写入到Elasticsearch5二、写入到Elasticsearch7总结 前言 Flink sink 流数据写入到es5和es7的简单示例。 一、写入到Elasticsearch5 pom maven依赖 <d…...

Java、Python、C++和C#的界面开发框架和工具的重新介绍

好的&#xff0c;以下是Java、Python、C和C#的界面开发框架和工具的重新介绍&#xff1a; Java界面开发&#xff1a; Swing: 是Java提供的一个基于组件的GUI工具包&#xff0c;可以创建跨平台的图形用户界面。它提供了丰富的组件和布局管理器&#xff0c;使得界面开发相对简单。…...

Java二叉树的遍历以及最大深度问题

Java学习面试指南&#xff1a;https://javaxiaobear.cn 1、树的相关概念 1、树的基本定义 树是我们计算机中非常重要的一种数据结构&#xff0c;同时使用树这种数据结构&#xff0c;可以描述现实生活中的很多事物&#xff0c;例如家谱、单位的组织架构、等等。 树是由n&#…...

Apollo 9.0搭建问题记录

虚拟机安装 可以看这个&#xff1a;https://blog.csdn.net/qq_45138078/article/details/129815408 写的很详细 内存 为了学习 Apollo &#xff0c;所以只是使用了虚拟机&#xff0c;内存得大一点&#xff08;128G&#xff09;&#xff0c;第一次&#xff0c;就是因为分配内…...

【心得】PHP文件包含高级利用攻击面个人笔记

目录 一、nginx日志文件包含 二、临时文件包含 三、php的session文件包含 四、pear文件包含 五 、远程文件包含 文件包含 include "/var/www/html/flag.php"; 一 文件名可控 $file$_GET[file]; include $file.".php"; //用php伪协议 &#xff0…...

[scala] 列表常见用法

文章目录 不可变列表 List可变列表 ListBuffer 不可变列表 List 在 Scala 中&#xff0c;列表是一种不可变的数据结构&#xff0c;用于存储一系列元素。列表使用 List 类来表示&#xff0c;它提供了许多方法来操作和处理列表。 下面是一些常见的使用列表的示例&#xff1a; 创…...

python 使用urllib3发起post请求,携带json参数

当通过python脚本&#xff0c;发起http post请求&#xff0c;网络上大多是通过fields传递数据&#xff0c;然而这样&#xff0c;服务器收到的请求&#xff0c;但无法解析json数据。类似这些链接&#xff1a; Python urllib3库使用指南 软件测试|Python urllib3库使用指南 p…...

深入理解堆(Heap):一个强大的数据结构

. 个人主页&#xff1a;晓风飞 专栏&#xff1a;数据结构|Linux|C语言 路漫漫其修远兮&#xff0c;吾将上下而求索 文章目录 前言堆的实现基本操作结构体定义初始化堆&#xff08;HeapInit&#xff09;销毁堆&#xff08;HeapDestroy&#xff09; 重要函数交换函数&#xff08;…...

抖音在线查权重系统源码,附带查询接口

抖音权重在线查询只需输入抖音主页链接&#xff0c;即可查询作品情况。 搭建教程 上传源码并解压 修改数据库“bygoukai.sql” 修改“config.php” 如需修改水印请修改第40行 如需修改限制次数&#xff0c;请修改第156行 访问域名user.php即可查看访问用户&#xff0c;停…...

Spring Framework和SpringBoot的区别

目录 一、前言 二、什么是Spring 三、什么是Spring Framework 四、什么是SpringBoot 五、使用Spring Framework构建工程 六、使用SpringBoot构建工程 七、总结 一、前言 作为Java程序员&#xff0c;我们都听说过Spring&#xff0c;也都使用过Spring的相关产品&#xff0…...

2024--Django平台开发-Django知识点(三)

day03 django知识点 项目相关路由相关 urls.py视图相关 views.py模版相关 templates资源相关 static/media 1.项目相关 新项目 开发时&#xff0c;可能遇到使用其他的版本。虚拟环境 老项目 打开项目虚拟环境 1.1 关于新项目 1.系统解释器命令行【学习】 C:/python38- p…...

Github 2024-01-08开源项目周报 Top14

根据Github Trendings的统计&#xff0c;本周(2024-01-08统计)共有14个项目上榜。根据开发语言中项目的数量&#xff0c;汇总情况如下&#xff1a; 开发语言项目数量Python项目5TypeScript项目3C项目2Dart项目1QML项目1Go项目1Shell项目1Rust项目1JavaScript项目1C#项目1 免费…...

vue3 的内置组件汇总

官方给出的说明&#xff1a; Fragment: Vue 3 组件不再要求有一个唯一的根节点&#xff0c;清除了很多无用的占位 div。Teleport: 允许组件渲染在别的元素内&#xff0c;主要开发弹窗组件的时候特别有用。Suspense: 异步组件&#xff0c;更方便开发有异步请求的组件。 一、fr…...

ARM工控机Node-red使用教程

嵌入式ARM工控机Node-red安装教程 从前车马很慢书信很远&#xff0c;而现在人们不停探索“科技改变生活”。 智能终端的出现改变了我们的生活方式&#xff0c;钡铼技术嵌入式工控机协助您灵活布建能源管理、大楼自动化、工业自动化、电动车充电站等各种多元性IoT应用&#xff…...

Visual Studio 发布程序自动更新 ClickOnce和AutoUpdater测试

文章目录 前言运行环境ClickOnce&#xff08;Visual Studio 程序发布&#xff09;IIS新建文件夹C# 控制台测试安装测试更新测试卸载 AutoUpdaterDotNET实现原理简单使用新建一个WPF项目 代码封装自动更新代码封装简单使用 总结 前言 虽然写的大部分都是不联网项目&#xff0c;…...

Codeforces Round 761 (Div. 2) E. Christmas Chocolates(思维题 树的直径 二进制性质 lca)

题目 n(n<2e5)个值&#xff0c;第i个值ai(0<ai<1e9)&#xff0c;所有ai两两不同 初始时&#xff0c;选择两个位置x,y(x≠y)&#xff0c;代表需要对这两个位置进行操作&#xff0c;要把其中一个值变成另一个 你可以执行若干次操作&#xff0c;每一次&#xff0c;你可…...

知识图谱之汽车实战案例综述与前瞻分析

知识图谱的前置介绍 什么是知识图谱 知识图谱本质(Knowledge Graph&#xff09;上是一种叫做语义网络(semantic network &#xff09; 的知识库&#xff0c;即具有有向图结构的一个知识库&#xff1b;图的结点代表实体&#xff08;entity&#xff09;或者概念&#xff08;con…...

网关Gateway

什么是网关? 网关实质上是一个网络通向其他网络的 IP 地址&#xff0c;是当前微服务项目的"统一入口"。 网关能做什么&#xff1f; 反向代理 、鉴权、 流量控制、 熔断、 日志监控等 图片原文&#xff1a;http://t.csdnimg.cn/SvUJh 核心概念 Router&#xff08;…...

java 生成一个当前时间的时间搓

开发过程中 用时间搓数值格式存储 会更加精准 那么 我们在一些日常增删查改中就可以用时间搓来记录操作时间 就一行代码 long timestamp System.currentTimeMillis();他就能生成当前时间的时间搓 运行结果如下 然后 我们可以在 http://shijianchuo.wiicha.com/ 上进行转换查…...

金融中IC和IR的定义

当谈到金融领域时&#xff0c;IC&#xff08;Information Coefficient&#xff09;和IR&#xff08;Information Ratio&#xff09;通常是用来评估投资组合管理绩效的指标。它们都涉及到投资者对信息的利用和管理的效果。 信息系数&#xff08;IC - Information Coefficient&a…...

React第五十七节 Router中RouterProvider使用详解及注意事项

前言 在 React Router v6.4 中&#xff0c;RouterProvider 是一个核心组件&#xff0c;用于提供基于数据路由&#xff08;data routers&#xff09;的新型路由方案。 它替代了传统的 <BrowserRouter>&#xff0c;支持更强大的数据加载和操作功能&#xff08;如 loader 和…...

postgresql|数据库|只读用户的创建和删除(备忘)

CREATE USER read_only WITH PASSWORD 密码 -- 连接到xxx数据库 \c xxx -- 授予对xxx数据库的只读权限 GRANT CONNECT ON DATABASE xxx TO read_only; GRANT USAGE ON SCHEMA public TO read_only; GRANT SELECT ON ALL TABLES IN SCHEMA public TO read_only; GRANT EXECUTE O…...

uniapp微信小程序视频实时流+pc端预览方案

方案类型技术实现是否免费优点缺点适用场景延迟范围开发复杂度​WebSocket图片帧​定时拍照Base64传输✅ 完全免费无需服务器 纯前端实现高延迟高流量 帧率极低个人demo测试 超低频监控500ms-2s⭐⭐​RTMP推流​TRTC/即构SDK推流❌ 付费方案 &#xff08;部分有免费额度&#x…...

C++.OpenGL (10/64)基础光照(Basic Lighting)

基础光照(Basic Lighting) 冯氏光照模型(Phong Lighting Model) #mermaid-svg-GLdskXwWINxNGHso {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-GLdskXwWINxNGHso .error-icon{fill:#552222;}#mermaid-svg-GLd…...

【Java_EE】Spring MVC

目录 Spring Web MVC ​编辑注解 RestController RequestMapping RequestParam RequestParam RequestBody PathVariable RequestPart 参数传递 注意事项 ​编辑参数重命名 RequestParam ​编辑​编辑传递集合 RequestParam 传递JSON数据 ​编辑RequestBody ​…...

ArcGIS Pro制作水平横向图例+多级标注

今天介绍下载ArcGIS Pro中如何设置水平横向图例。 之前我们介绍了ArcGIS的横向图例制作&#xff1a;ArcGIS横向、多列图例、顺序重排、符号居中、批量更改图例符号等等&#xff08;ArcGIS出图图例8大技巧&#xff09;&#xff0c;那这次我们看看ArcGIS Pro如何更加快捷的操作。…...

重启Eureka集群中的节点,对已经注册的服务有什么影响

先看答案&#xff0c;如果正确地操作&#xff0c;重启Eureka集群中的节点&#xff0c;对已经注册的服务影响非常小&#xff0c;甚至可以做到无感知。 但如果操作不当&#xff0c;可能会引发短暂的服务发现问题。 下面我们从Eureka的核心工作原理来详细分析这个问题。 Eureka的…...

蓝桥杯 冶炼金属

原题目链接 &#x1f527; 冶炼金属转换率推测题解 &#x1f4dc; 原题描述 小蓝有一个神奇的炉子用于将普通金属 O O O 冶炼成为一种特殊金属 X X X。这个炉子有一个属性叫转换率 V V V&#xff0c;是一个正整数&#xff0c;表示每 V V V 个普通金属 O O O 可以冶炼出 …...

Mysql中select查询语句的执行过程

目录 1、介绍 1.1、组件介绍 1.2、Sql执行顺序 2、执行流程 2.1. 连接与认证 2.2. 查询缓存 2.3. 语法解析&#xff08;Parser&#xff09; 2.4、执行sql 1. 预处理&#xff08;Preprocessor&#xff09; 2. 查询优化器&#xff08;Optimizer&#xff09; 3. 执行器…...

GitHub 趋势日报 (2025年06月06日)

&#x1f4ca; 由 TrendForge 系统生成 | &#x1f310; https://trendforge.devlive.org/ &#x1f310; 本日报中的项目描述已自动翻译为中文 &#x1f4c8; 今日获星趋势图 今日获星趋势图 590 cognee 551 onlook 399 project-based-learning 348 build-your-own-x 320 ne…...