• SpringData、SparkStreaming和Flink集成Elasticsearch

    本文代码链接：https://download.csdn.net/download/shangjg03/88522188

    1 Spring Data框架集成

    1.1 Spring Data框架介绍

    Spring Data是一个用于简化数据库、非关系型数据库、索引库访问，并支持云服务的开源框架。其主要目标是使得对数据的访问变得方便快捷，并支持map-reduce框架和云计算数据服务。 Spring Data可以极大的简化JPA（Elasticsearch…）的写法，可以在几乎不用写实现的情况下，实现对数据的访问和操作。除了CRUD外，还包括如分页、排序等一些常用的功能。

    Spring Data的官网：Spring Data

    Spring Data常用的功能模块如下：

    1.2 Spring Data Elasticsearch介绍

    Spring Data Elasticsearch 基于 spring data API 简化 Elasticsearch操作，将原始操作Elasticsearch的客户端API 进行封装 。Spring Data为Elasticsearch项目提供集成搜索引擎。Spring Data Elasticsearch POJO的关键功能区域为中心的模型与Elastichsearch交互文档和轻松地编写一个存储索引库数据访问层。

    官方网站: https://spring.io/projects/spring-data-elasticsearch

    1.3 Spring Data Elasticsearch版本对比

    目前最新springboot对应Elasticsearch7.6.2，Spring boot2.3.x一般可以兼容Elasticsearch7.x

    1.4 框架集成

    1. 创建Maven项目

    1. 修改pom文件，增加依赖关系

    1. <?xml version="1.0" encoding="UTF-8"?>
    2. <project xmlns="http://maven.apache.org/POM/4.0.0"
    3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.     <modelVersion>4.0.0</modelVersion>
    6.  
    7.     <parent>
    8.         <groupId>org.springframework.boot</groupId>
    9.         <artifactId>spring-boot-starter-parent</artifactId>
    10.         <version>2.3.6.RELEASE</version>
    11.         <relativePath/>
    12.     </parent>
    13.  
    14.  
    15.     <groupId>com.shangjack.es</groupId>
    16.     <artifactId>springdata-elasticsearch</artifactId>
    17.     <version>1.0</version>
    18.  
    19.     <properties>
    20.         <maven.compiler.source>8</maven.compiler.source>
    21.         <maven.compiler.target>8</maven.compiler.target>
    22.     </properties>
    23.  
    24.     <dependencies>
    25.         <dependency>
    26.             <groupId>org.projectlombok</groupId>
    27.             <artifactId>lombok</artifactId>
    28.         </dependency>
    29.  
    30.         <dependency>
    31.             <groupId>org.springframework.boot</groupId>
    32.             <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
    33.         </dependency>
    34.  
    35.         <dependency>
    36.             <groupId>org.springframework.boot</groupId>
    37.             <artifactId>spring-boot-devtools</artifactId>
    38.             <scope>runtime</scope>
    39.             <optional>true</optional>
    40.         </dependency>
    41.         <dependency>
    42.             <groupId>org.springframework.boot</groupId>
    43.             <artifactId>spring-boot-starter-test</artifactId>
    44.             <scope>test</scope>
    45.         </dependency>
    46.         <dependency>
    47.             <groupId>org.springframework.boot</groupId>
    48.             <artifactId>spring-boot-test</artifactId>
    49.         </dependency>
    50.         <dependency>
    51.             <groupId>junit</groupId>
    52.             <artifactId>junit</artifactId>
    53.         </dependency>
    54.         <dependency>
    55.             <groupId>org.springframework</groupId>
    56.             <artifactId>spring-test</artifactId>
    57.         </dependency>
    58.     </dependencies>
    59. </project>
    60.

    1. 增加配置文件

    在resources目录中增加application.properties文件

    1. # es服务地址
    2. elasticsearch.host=127.0.0.1
    3. # es服务端口
    4. elasticsearch.port=9200
    5. # 配置日志级别,开启debug日志
    6. logging.level.com.shangjack.es=debug

    1. SpringBoot主程序

    package com.shangjack.es;

    1.  
    2. import org.springframework.boot.SpringApplication;
    3. import org.springframework.boot.autoconfigure.SpringBootApplication;
    4.  
    5. @SpringBootApplication
    6. public class SpringDataElasticSearchMainApplication {
    7.     public static void main(String[] args) {
    8.         SpringApplication.run(SpringDataElasticSearchMainApplication.class,args);
    9.     }
    10. }
    11. 数据实体类
    12. package com.shangjack.es;
    13.  
    14. import lombok.AllArgsConstructor;
    15. import lombok.Data;
    16. import lombok.NoArgsConstructor;
    17. import lombok.ToString;
    18.  
    19. @Data
    20. @NoArgsConstructor
    21. @AllArgsConstructor
    22. @ToString
    23. public class Product {
    24.     private Long id;//商品唯一标识
    25.     private String title;//商品名称
    26.     private String category;//分类名称
    27.     private Double price;//商品价格
    28.     private String images;//图片地址
    29.  
    30. }

    1. 配置类
    • ElasticsearchRestTemplate是spring-data-elasticsearch项目中的一个类，和其他spring项目中的template类似。
    • 在新版的spring-data-elasticsearch中，ElasticsearchRestTemplate代替了原来的ElasticsearchTemplate。
    • 原因是ElasticsearchTemplate基于TransportClient，TransportClient即将在8.x以后的版本中移除。所以，我们推荐使用ElasticsearchRestTemplate。
    • ElasticsearchRestTemplate基于RestHighLevelClient客户端的。需要自定义配置类，继承AbstractElasticsearchConfiguration，并实现elasticsearchClient()抽象方法，创建RestHighLevelClient对象。

    1. package com.shangjack.es;
    2.  
    3. import lombok.Data;
    4. import org.apache.http.HttpHost;
    5. import org.elasticsearch.client.RestClient;
    6. import org.elasticsearch.client.RestClientBuilder;
    7. import org.elasticsearch.client.RestHighLevelClient;
    8. import org.springframework.boot.context.properties.ConfigurationProperties;
    9. import org.springframework.context.annotation.Configuration;
    10. import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;
    11.  
    12. @ConfigurationProperties(prefix = "elasticsearch")
    13. @Configuration
    14. @Data
    15. public class ElasticsearchConfig extends AbstractElasticsearchConfiguration {
    16.     private String host ;
    17.     private Integer port ;
    18.  
    19.     //重写父类方法
    20.     @Override
    21.     public RestHighLevelClient elasticsearchClient() {
    22.         RestClientBuilder builder = RestClient.builder(new HttpHost(host, port));
    23.         RestHighLevelClient restHighLevelClient = new RestHighLevelClient(builder);
    24.         return restHighLevelClient;
    25.     }
    26. }

    1. DAO数据访问对象

    1. package com.shangjack.es;
    2.  
    3. import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
    4. import org.springframework.stereotype.Repository;
    5.  
    6. @Repository
    7. public interface ProductDao extends ElasticsearchRepository<Product,Long> {
    8.  
    9. }

    1. 实体类映射操作

    1. package com.shangjack.es;
    2.  
    3. import lombok.AllArgsConstructor;
    4. import lombok.Data;
    5. import lombok.NoArgsConstructor;
    6. import lombok.ToString;
    7. import org.springframework.data.annotation.Id;
    8. import org.springframework.data.elasticsearch.annotations.Document;
    9. import org.springframework.data.elasticsearch.annotations.Field;
    10. import org.springframework.data.elasticsearch.annotations.FieldType;
    11.  
    12. @Data
    13. @NoArgsConstructor
    14. @AllArgsConstructor
    15. @ToString
    16. @Document(indexName = "shopping", shards = 3, replicas = 1)
    17. public class Product {
    18.     //必须有id,这里的id是全局唯一的标识，等同于es中的"_id"
    19.     @Id
    20.     private Long id;//商品唯一标识
    21.     /**
    22.      * type : 字段数据类型
    23.      * analyzer : 分词器类型
    24.      * index : 是否索引(默认:true)
    25.      * Keyword : 短语,不进行分词
    26.      */
    28.     @Field(type = FieldType.Text, analyzer = "ik_max_word")
    29.     private String title;//商品名称
    30.     @Field(type = FieldType.Keyword)
    31.     private String category;//分类名称
    32.     @Field(type = FieldType.Double)
    33.     private Double price;//商品价格
    34.     @Field(type = FieldType.Keyword, index = false)
    35.     private String images;//图片地址
    36. }

    1. 索引操作

    1. package com.shangjack.es;
    2.  
    3. import org.junit.Test;
    4. import org.junit.runner.RunWith;
    5. import org.springframework.beans.factory.annotation.Autowired;
    6. import org.springframework.boot.test.context.SpringBootTest;
    7. import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
    8. import org.springframework.test.context.junit4.SpringRunner;
    9.  
    10. @RunWith(SpringRunner.class)
    11. @SpringBootTest
    12. public class SpringDataESIndexTest {
    13.     //注入ElasticsearchRestTemplate
    14.     @Autowired
    15.     private ElasticsearchRestTemplate elasticsearchRestTemplate;
    16.  
    17.     //创建索引并增加映射配置
    18.     @Test
    19.     public void createIndex(){
    20.         //创建索引，系统初始化会自动创建索引
    21.         System.out.println("创建索引");
    22.     }
    23.  
    24.     @Test
    25.     public void deleteIndex(){
    26.         //创建索引，系统初始化会自动创建索引
    27.         boolean flg = elasticsearchRestTemplate.deleteIndex(Product.class);
    28.         System.out.println("删除索引 = " + flg);
    29.     }
    30. }

    1. 文档操作

    1. package com.shangjack.es;
    2.  
    3. import org.junit.Test;
    4. import org.junit.runner.RunWith;
    5. import org.springframework.beans.factory.annotation.Autowired;
    6. import org.springframework.boot.test.context.SpringBootTest;
    7. import org.springframework.data.domain.Page;
    8. import org.springframework.data.domain.PageRequest;
    9. import org.springframework.data.domain.Sort;
    10. import org.springframework.test.context.junit4.SpringRunner;
    11.  
    12. import java.util.ArrayList;
    13. import java.util.List;
    14.  
    15. @RunWith(SpringRunner.class)
    16. @SpringBootTest
    17. public class SpringDataESProductDaoTest {
    18.     @Autowired
    19.     private ProductDao productDao;
    20.  
    21.     /**
    22.      * 新增
    23.      */
    24.     @Test
    25.     public void save(){
    26.         Product product = new Product();
    27.         product.setId(2L);
    28.         product.setTitle("华为手机");
    29.         product.setCategory("手机");
    30.         product.setPrice(2999.0);
    31.         product.setImages("http://www.shangjack/hw.jpg");
    32.         productDao.save(product);
    33.     }
    34.  
    35.     //修改
    36.     @Test
    37.     public void update(){
    38.         Product product = new Product();
    39.         product.setId(1L);
    40.         product.setTitle("小米2手机");
    41.         product.setCategory("手机");
    42.         product.setPrice(9999.0);
    43.         product.setImages("http://www.shangjack/xm.jpg");
    44.         productDao.save(product);
    45.     }
    46.  
    47.     //根据id查询
    48.     @Test
    49.     public void findById(){
    50.         Product product = productDao.findById(1L).get();
    51.         System.out.println(product);
    52.     }
    53.  
    54.     //查询所有
    55.     @Test
    56.     public void findAll(){
    57.         Iterable<Product> products = productDao.findAll();
    58.         for (Product product : products) {
    59.             System.out.println(product);
    60.         }
    61.     }
    62.  
    63.     //删除
    64.     @Test
    65.     public void delete(){
    66.         Product product = new Product();
    67.         product.setId(1L);
    68.         productDao.delete(product);
    69.     }
    70.  
    71.     //批量新增
    72.     @Test
    73.     public void saveAll(){
    74.         List<Product> productList = new ArrayList<>();
    75.         for (int i = 0; i < 10; i++) {
    76.             Product product = new Product();
    77.             product.setId(Long.valueOf(i));
    78.             product.setTitle("["+i+"]小米手机");
    79.             product.setCategory("手机");
    80.             product.setPrice(1999.0+i);
    81.             product.setImages("http://www.shangjack/xm.jpg");
    82.             productList.add(product);
    83.         }
    84.         productDao.saveAll(productList);
    85.     }
    86.  
    87.     //分页查询
    88.     @Test
    89.     public void findByPageable(){
    90.         //设置排序(排序方式，正序还是倒序，排序的id)
    91.         Sort sort = Sort.by(Sort.Direction.DESC,"id");
    92.         int currentPage=0;//当前页，第一页从0开始，1表示第二页
    93.         int pageSize = 5;//每页显示多少条
    94.         //设置查询分页
    95.         PageRequest pageRequest = PageRequest.of(currentPage, pageSize,sort);
    96.         //分页查询
    97.         Page<Product> productPage = productDao.findAll(pageRequest);
    98.         for (Product Product : productPage.getContent()) {
    99.             System.out.println(Product);
    100.         }
    101.     }
    102. }

    1. 文档搜索

    1. package com.shangjack.es;
    2.  
    3. import org.elasticsearch.index.query.QueryBuilders;
    4. import org.elasticsearch.index.query.TermQueryBuilder;
    5. import org.junit.Test;
    6. import org.junit.runner.RunWith;
    7. import org.springframework.beans.factory.annotation.Autowired;
    8. import org.springframework.boot.test.context.SpringBootTest;
    9. import org.springframework.data.domain.PageRequest;
    10. import org.springframework.test.context.junit4.SpringRunner;
    11.  
    12. @RunWith(SpringRunner.class)
    13. @SpringBootTest
    14. public class SpringDataESSearchTest {
    15.     @Autowired
    16.     private ProductDao productDao;
    17.  
    18.     /**
    19.      * term查询
    20.      * search(termQueryBuilder) 调用搜索方法，参数查询构建器对象
    21.      */
    22.     @Test
    23.     public void termQuery(){
    24.         TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("title", "小米");
    25.         Iterable<Product> products = productDao.search(termQueryBuilder);
    26.         for (Product product : products) {
    27.             System.out.println(product);
    28.         }
    29.     }
    30.  
    31.     /**
    32.      * term查询加分页
    33.      */
    34.     @Test
    35.     public void termQueryByPage(){
    36.         int currentPage= 0 ;
    37.         int pageSize = 5;
    38.         //设置查询分页
    39.         PageRequest pageRequest = PageRequest.of(currentPage, pageSize);
    40.         TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("title", "小米");
    41.         Iterable<Product> products = productDao.search(termQueryBuilder,pageRequest);
    42.         for (Product product : products) {
    43.             System.out.println(product);
    44.         }
    45.     }
    46.  
    47. }

    2 Spark Streaming框架集成

    2.1 Spark Streaming框架介绍

    Spark Streaming是Spark core API的扩展，支持实时数据流的处理，并且具有可扩展，高吞吐量，容错的特点。 数据可以从许多来源获取，如Kafka，Flume，Kinesis或TCP sockets，并且可以使用复杂的算法进行处理，这些算法使用诸如map，reduce，join和window等高级函数表示。 最后，处理后的数据可以推送到文件系统，数据库等。 实际上，您可以将Spark的机器学习和图形处理算法应用于数据流。

    2.2 框架集成

    1. 创建Maven项目

    1. 修改pom文件，增加依赖关系

    1. <?xml version="1.0" encoding="UTF-8"?>
    2. <project xmlns="http://maven.apache.org/POM/4.0.0"
    3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.     <modelVersion>4.0.0</modelVersion>
    6.  
    7.     <groupId>com.shangjack.es</groupId>
    8.     <artifactId>sparkstreaming-elasticsearch</artifactId>
    9.     <version>1.0</version>
    10.  
    11.     <properties>
    12.         <maven.compiler.source>8</maven.compiler.source>
    13.         <maven.compiler.target>8</maven.compiler.target>
    14.     </properties>
    15.  
    16.     <dependencies>
    17.         <dependency>
    18.             <groupId>org.apache.spark</groupId>
    19.             <artifactId>spark-core_2.12</artifactId>
    20.             <version>3.0.0</version>
    21.         </dependency>
    22.         <dependency>
    23.             <groupId>org.apache.spark</groupId>
    24.             <artifactId>spark-streaming_2.12</artifactId>
    25.             <version>3.0.0</version>
    26.         </dependency>
    27.         <dependency>
    28.             <groupId>org.elasticsearch</groupId>
    29.             <artifactId>elasticsearch</artifactId>
    30.             <version>7.8.0</version>
    31.         </dependency>
    32.         <!-- elasticsearch的客户端 -->
    33.         <dependency>
    34.             <groupId>org.elasticsearch.client</groupId>
    35.             <artifactId>elasticsearch-rest-high-level-client</artifactId>
    36.             <version>7.8.0</version>
    37.         </dependency>
    38.         <!-- elasticsearch依赖2.x的log4j -->
    39.         <dependency>
    40.             <groupId>org.apache.logging.log4j</groupId>
    41.             <artifactId>log4j-api</artifactId>
    42.             <version>2.8.2</version>
    43.         </dependency>
    44.         <dependency>
    45.             <groupId>org.apache.logging.log4j</groupId>
    46.             <artifactId>log4j-core</artifactId>
    47.             <version>2.8.2</version>
    48.         </dependency>
    49. <!--        <dependency>-->
    50. <!--            <groupId>com.fasterxml.jackson.core</groupId>-->
    51. <!--            <artifactId>jackson-databind</artifactId>-->
    52. <!--            <version>2.11.1</version>-->
    53. <!--        </dependency>-->
    54. <!--        &lt;!&ndash; junit单元测试 &ndash;&gt;-->
    55. <!--        <dependency>-->
    56. <!--            <groupId>junit</groupId>-->
    57. <!--            <artifactId>junit</artifactId>-->
    58. <!--            <version>4.12</version>-->
    59. <!--        </dependency>-->
    60.     </dependencies>
    61. </project>
    62.

    1. 功能实现

    1. package com.shangjack.es
    2.  
    3. import org.apache.http.HttpHost
    4. import org.apache.spark.SparkConf
    5. import org.apache.spark.streaming.dstream.ReceiverInputDStream
    6. import org.apache.spark.streaming.{Seconds, StreamingContext}
    7. import org.elasticsearch.action.index.IndexRequest
    8. import org.elasticsearch.client.indices.CreateIndexRequest
    9. import org.elasticsearch.client.{RequestOptions, RestClient, RestHighLevelClient}
    10. import org.elasticsearch.common.xcontent.XContentType
    11.  
    12. import java.util.Date
    13.  
    14. object SparkStreamingESTest {
    15.  
    16.     def main(args: Array[String]): Unit = {
    17.         val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ESTest")
    18.         val ssc = new StreamingContext(sparkConf, Seconds(3))
    19.  
    20.         val ds: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    21.  
    22.         ds.foreachRDD(
    23.             rdd => {
    24.                 println("*************** " + new Date())
    25.                 rdd.foreach(
    26.                     data => {
    27.                         val client = new RestHighLevelClient(
    28.                             RestClient.builder(new HttpHost("localhost", 9200, "http"))
    29.                         );
    30.                         // 新增文档 - 请求对象
    31.                         val request = new IndexRequest();
    32.                         // 设置索引及唯一性标识
    33.                         val ss = data.split(" ")
    34.                         println("ss = " + ss.mkString(","))
    35.                         request.index("sparkstreaming").id(ss(0));
    36.                         val productJson =
    37.                             s"""
    38.                               | { "data":"${ss(1)}" }
    39.                               |""".stripMargin;
    40.                         // 添加文档数据，数据格式为JSON格式
    41.                         request.source(productJson,XContentType.JSON);
    42.                         // 客户端发送请求，获取响应对象
    43.                         val response = client.index(request, RequestOptions.DEFAULT);
    44.                         System.out.println("_index:" + response.getIndex());
    45.                         System.out.println("_id:" + response.getId());
    46.                         System.out.println("_result:" + response.getResult());
    47.  
    48.                         client.close()
    49.                     }
    50.                 )
    51.             }
    52.         )
    53.  
    54.         ssc.start()
    55.         ssc.awaitTermination()
    56.     }
    57. }

    3 Flink框架集成

    3.1 Flink框架介绍

    Apache Spark是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。

    Apache Spark掀开了内存计算的先河，以内存作为赌注，赢得了内存计算的飞速发展。但是在其火热的同时，开发人员发现，在Spark中，计算框架普遍存在的缺点和不足依然没有完全解决，而这些问题随着5G时代的来临以及决策者对实时数据分析结果的迫切需要而凸显的更加明显：

    • 数据精准一次性处理（Exactly-Once）
    • 乱序数据，迟到数据
    • 低延迟，高吞吐，准确性
    • 容错性

    Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。在Spark火热的同时，也默默地发展自己，并尝试着解决其他计算框架的问题。

    慢慢地，随着这些问题的解决，Flink慢慢被绝大数程序员所熟知并进行大力推广，阿里公司在2015年改进Flink，并创建了内部分支Blink，目前服务于阿里集团内部搜索、推荐、广告和蚂蚁等大量核心实时业务。

    3.2 框架集成

    1. 创建Maven项目

    1. 修改pom文件，增加相关依赖类库

    1. <?xml version="1.0" encoding="UTF-8"?>
    2. <project xmlns="http://maven.apache.org/POM/4.0.0"
    3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.     <modelVersion>4.0.0</modelVersion>
    6.  
    7.     <groupId>com.shangjack.es</groupId>
    8.     <artifactId>flink-elasticsearch</artifactId>
    9.     <version>1.0</version>
    10.  
    11.     <properties>
    12.         <maven.compiler.source>8</maven.compiler.source>
    13.         <maven.compiler.target>8</maven.compiler.target>
    14.     </properties>
    15.  
    16.     <dependencies>
    17.         <dependency>
    18.             <groupId>org.apache.flink</groupId>
    19.             <artifactId>flink-scala_2.12</artifactId>
    20.             <version>1.12.0</version>
    21.         </dependency>
    22.         <dependency>
    23.             <groupId>org.apache.flink</groupId>
    24.             <artifactId>flink-streaming-scala_2.12</artifactId>
    25.             <version>1.12.0</version>
    26.         </dependency>
    27.         <dependency>
    28.             <groupId>org.apache.flink</groupId>
    29.             <artifactId>flink-clients_2.12</artifactId>
    30.             <version>1.12.0</version>
    31.         </dependency>
    32.  
    33.         <dependency>
    34.             <groupId>org.apache.flink</groupId>
    35.             <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
    36.             <version>1.12.0</version>
    37.         </dependency>
    38.  
    39.         <!-- jackson -->
    40.         <dependency>
    41.             <groupId>com.fasterxml.jackson.core</groupId>
    42.             <artifactId>jackson-core</artifactId>
    43.             <version>2.11.1</version>
    44.         </dependency>
    45.     </dependencies>
    46. </project>

    1. 功能实现

    1. package com.shangjack.es;
    2.  
    3. import org.apache.flink.api.common.functions.RuntimeContext;
    4. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    6. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
    7. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
    8. import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
    9. import org.apache.http.HttpHost;
    10. import org.elasticsearch.action.index.IndexRequest;
    11. import org.elasticsearch.client.Requests;
    12.  
    13. import java.util.ArrayList;
    14. import java.util.HashMap;
    15. import java.util.List;
    16. import java.util.Map;
    17.  
    18. public class FlinkElasticsearchSinkTest {
    19.     public static void main(String[] args) throws Exception {
    20.  
    21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    22.  
    23.         DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
    24.  
    25.         List<HttpHost> httpHosts = new ArrayList<>();
    26.         httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
    27.         //httpHosts.add(new HttpHost("10.2.3.1", 9200, "http"));
    28.  
    29. // use a ElasticsearchSink.Builder to create an ElasticsearchSink
    30.         ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
    31.                 httpHosts,
    32.                 new ElasticsearchSinkFunction<String>() {
    33.                     public IndexRequest createIndexRequest(String element) {
    34.                         Map<String, String> json = new HashMap<>();
    35.                         json.put("data", element);
    36.  
    37.                         return Requests.indexRequest()
    38.                                 .index("my-index")
    39.                                 //.type("my-type")
    40.                                 .source(json);
    41.                     }
    42.  
    43.                     @Override
    44.                     public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
    45.                         indexer.add(createIndexRequest(element));
    46.                     }
    47.                 }
    48.         );
    49.  
    50. // configuration for the bulk requests; this instructs the sink to emit after every element, otherwise they would be buffered
    51.         esSinkBuilder.setBulkFlushMaxActions(1);
    52.  
    53. // provide a RestClientFactory for custom configuration on the internally created REST client
    54. //        esSinkBuilder.setRestClientFactory(
    55. //                restClientBuilder -> {
    56. //                    restClientBuilder.setDefaultHeaders(...)
    57. //                    restClientBuilder.setMaxRetryTimeoutMillis(...)
    58. //                    restClientBuilder.setPathPrefix(...)
    59. //                    restClientBuilder.setHttpClientConfigCallback(...)
    60. //                }
    61. //        );
    62.  
    63.         source.addSink(esSinkBuilder.build());
    64.  
    65.         env.execute("flink-es");
    66.     }
    67. }

  • 相关阅读:
    The 2022 ICPC Asia Xian Regional Contest--C. Clone Ranran
    数据可视化系列教程之组件构成
    面试题：CSS 怎样实现动画？
    攻防世界数据逆向 2023
    网络系统管理 - GWServer虚拟机配置
    信号的处理时机(内核态,用户态,如何/为什么相互转换,内核空间,cpu寄存器),信号的处理流程详细介绍+抽象图解
    L1-023 输出GPLT C++解法【全网最细讲解】
    Clickhouse：clickhouse切换目录
    深度解读《深度探索C++对象模型》之数据成员的存取效率分析（一）
    基于昇腾AI 使用AscendCL实现垃圾分类和视频物体分类应用
  • 原文地址：https://blog.csdn.net/shangjg03/article/details/134336259