devbox
小丑西瓜9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

flink将数据结果写入elasticsearch_flink写入es

作者：小丑西瓜9 | 2024-05-22 04:37:42

赞

踩

flink写入es

一、需求

用flink流处理，将结果数据落地到elasticsearch中。

二、软件版本

flink1.8.0

elasticsearch7.8.1

kafka_2.11-1.0.0

java1.8

三、代码

1、maven的pom文件

  1. <properties>
  2.     <compiler.version>1.8</compiler.version>
  3.     <flink.version>1.8.0</flink.version>
  4.     <java.version>1.8</java.version>
  5.     <hadoop.version>2.7.5</hadoop.version>
  6.     <scala.binary.version>2.11</scala.binary.version>
  7.     <maven.compiler.source>1.8</maven.compiler.source>
  8.     <maven.compiler.target>1.8</maven.compiler.target>
  9.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  10. </properties>
  11.  
  12. <dependencies>
  13.     <dependency>
  14.         <groupId>org.apache.flink</groupId>
  15.         <artifactId>flink-connector-kafka-0.11_${scala.binary.version}</artifactId>
  16.         <version>${flink.version}</version>
  17.     </dependency>
  18.     <dependency>
  19.         <groupId>org.apache.flink</groupId>
  20.         <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  21.         <version>${flink.version}</version>
  22.         <scope>provided</scope>
  23.     </dependency>
  24.     <!--加入下面两个依赖才会出现 Flink 的日志出来-->
  25.     <dependency>
  26.         <groupId>org.slf4j</groupId>
  27.         <artifactId>slf4j-api</artifactId>
  28.         <version>1.7.25</version>
  29.     </dependency>
  30.     <dependency>
  31.         <groupId>org.slf4j</groupId>
  32.         <artifactId>slf4j-simple</artifactId>
  33.         <version>1.7.25</version>
  34.     </dependency>
  35.     <!--对象和json 互相转换的-->
  36.     <dependency>
  37.         <groupId>com.alibaba</groupId>
  38.         <artifactId>fastjson</artifactId>
  39.         <version>1.2.44</version>
  40.     </dependency>
  41.     <dependency>
  42.         <groupId>org.apache.hadoop</groupId>
  43.         <artifactId>hadoop-client</artifactId>
  44.         <version>${hadoop.version}</version>
  45.     </dependency>
  46.     <!-- 指定mysql-connector的依赖 -->
  47.     <dependency>
  48.         <groupId>mysql</groupId>
  49.         <artifactId>mysql-connector-java</artifactId>
  50.         <version>5.1.38</version>
  51.     </dependency>
  52.     <dependency>
  53.         <groupId>org.apache.flink</groupId>
  54.         <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
  55.         <version>1.8.0</version>
  56.     </dependency>
  57.  
  58.     <dependency>
  59.         <groupId>org.apache.httpcomponents</groupId>
  60.         <artifactId>httpclient</artifactId>
  61.         <version>4.5.5</version>
  62.     </dependency>
  63. </dependencies>
  64.  
  65. <!-- This profile helps to make things run out of the box in IntelliJ -->
  66. <!-- Its adds Flink's core classes to the runtime class path. -->
  67. <!-- Otherwise they are missing in IntelliJ, because the dependency is 'provided' -->
  68. <profiles>
  69.     <profile>
  70.         <id>add-dependencies-for-IDEA</id>
  71.  
  72.         <activation>
  73.             <property>
  74.                 <name>idea.version</name>
  75.             </property>
  76.         </activation>
  77.  
  78.         <dependencies>
  79.             <dependency>
  80.                 <groupId>org.apache.flink</groupId>
  81.                 <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  82.                 <version>${flink.version}</version>
  83.                 <scope>compile</scope>
  84.             </dependency>
  85.         </dependencies>
  86.     </profile>
  87. </profiles>

2、代码

  1. package com.felix.kafka;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.felix.model.SecureLog;
  6. import com.felix.utils.CommonUtils;
  7. import com.felix.utils.DateUtils;
  8. import com.felix.utils.HttpUtils;
  9. import com.felix.utils.PropertyUtils;
  10. import org.apache.flink.api.common.functions.MapFunction;
  11. import org.apache.flink.api.common.functions.RuntimeContext;
  12. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  13. import org.apache.flink.streaming.api.TimeCharacteristic;
  14. import org.apache.flink.streaming.api.datastream.DataStream;
  15. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  16. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  17. import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
  18. import org.apache.flink.streaming.api.watermark.Watermark;
  19. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
  20. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
  21. import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
  22. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
  23. import org.apache.http.Header;
  24. import org.apache.http.HttpHost;
  25. import org.apache.http.message.BasicHeader;
  26. import org.elasticsearch.action.index.IndexRequest;
  27. import org.elasticsearch.client.Requests;
  28.  
  29. import javax.annotation.Nullable;
  30. import java.util.*;
  31.  
  32. public class KafkaMain {
  33.  
  34.     public static void main(String[] args) throws Exception{
  35.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  36.  
  37.         //设置容错机制
  38.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  39.         env.setParallelism(1);
  40.         // 5s钟启动一次checkpoint   Maven projects need to be imported: Import Changes   Enable Auto-Import
  41.  
  42. //        env.enableCheckpointing(10000);
  43. //
  44. //        // 设置checkpoint只checkpoint一次
  45. //        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  46. //        // 设置两次checkpoint的最小时间间隔
  47. //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
  48. //        // 设置checkpoint的超时时长
  49. //        env.getCheckpointConfig().setCheckpointTimeout(60000);
  50. //        // 最大并行度
  51. //        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  52. //        // 当程序关闭时,触发额外的checkpoint
  53. //        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  54. //
  55. //        // 设置checkpoint的地址
  56. //        env.setStateBackend(new FsStateBackend("hdfs://node01:8020/flink-checkpoint/"));
  57.  
  58.  
  59.         Properties props = PropertyUtils.getPropertiesWithKafkaConfig();
  60.         DataStreamSource<String> dataStreamSource = env.addSource(new FlinkKafkaConsumer011<>(
  61.                 PropertyUtils.get("input.topic"),  //kafka topic
  62.                 new SimpleStringSchema(),  // String 序列化
  63.                 props)).setParallelism(1);
  64.  
  65.      //   dataStreamSource.print(); //把从 kafka 读取到的数据打印在控制台
  66.  
  67.         DataStream<SecureLog> secureLogStream = dataStreamSource.map(new MapFunction<String, SecureLog>() {
  68.             @Override
  69.             public SecureLog map(String msgJson) throws Exception {
  70.                 SecureLog secureLog = new SecureLog();
  71.                 JSONObject jsonObject = JSON.parseObject(msgJson);
  72.                 secureLog.setSyslogProgram(jsonObject.getString("syslog_program"));
  73.                 secureLog.setPath(jsonObject.getString("path"));
  74.                 secureLog.setSyslogTimestamp(jsonObject.getString("syslog_timestamp"));
  75.                 secureLog.setTimestamp(jsonObject.getString("@timestamp"));
  76.                 secureLog.setSyslogHostname(jsonObject.getString("syslog_hostname"));
  77.                 secureLog.setVersion(jsonObject.getString("@version"));
  78.                 secureLog.setMessage(jsonObject.getString("message"));
  79.                 secureLog.setHost(jsonObject.getString("host"));
  80.                 secureLog.setReceivedFrom(jsonObject.getString("received_from"));
  81.                 secureLog.setSyslogPid(jsonObject.getString("syslog_pid"));
  82.                 secureLog.setReceivedAt(jsonObject.getString("received_at"));
  83.                 secureLog.setSyslogMessage(jsonObject.getString("syslog_message"));
  84.                 secureLog.setType(jsonObject.getString("type"));
  85.                 secureLog.setSysLogTimestampLong(DateUtils.dateStrToLong(secureLog.getSyslogTimestamp()));
  86.                 return secureLog;
  87.             }
  88.         });
  89.  
  90. //        secureLogStream.print();
  91.  
  92.         //设置水印机制
  93.         DataStream<SecureLog> secureLogDataStreamWithWaterMask = secureLogStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<SecureLog>() {
  94.             Long currentTimeStamp = 0l;
  95.             // 延迟时间
  96.             Long maxDelayTime = 2000l;
  97.             @Nullable
  98.             @Override
  99.             public Watermark getCurrentWatermark() {
  100.                 return new Watermark(currentTimeStamp - maxDelayTime);
  101.             }
  102.  
  103.             @Override
  104.             public long extractTimestamp(SecureLog secureLog, long previousElementTimestamp) {
  105.  
  106.                 return Math.max(secureLog.getSysLogTimestampLong(), previousElementTimestamp);
  107.             }
  108.         });
  109.  
  110. //        secureLogDataStreamWithWaterMask.print();
  111.  
  112.  
  113.         //落地到mysql
  114. //        secureLogDataStreamWithWaterMask.addSink(new MySqlSink());
  115.  
  116.  
  117.         //落地到es
  118.         List<HttpHost> httpHosts = new ArrayList<>();
  119.         httpHosts.add(new HttpHost("192.168.126.128", 9200, "http"));
  120.         httpHosts.add(new HttpHost("192.168.126.129", 9200, "http"));
  121.         httpHosts.add(new HttpHost("192.168.126.130", 9200, "http"));
  122.  
  123.  
  124.         // use a ElasticsearchSink.Builder to create an ElasticsearchSink
  125.         ElasticsearchSink.Builder<SecureLog> esSinkBuilder = new ElasticsearchSink.Builder<SecureLog>(
  126.                 httpHosts,
  127.                 new ElasticsearchSinkFunction<SecureLog>() {
  128.                     public IndexRequest createIndexRequest(SecureLog secureLog) {
  129. //                        Map<String, SecureLog> json = new HashMap<>();
  130. //                        json.put("data", secureLog);
  131.                         Map<String, Object> map = CommonUtils.beanToMap(secureLog);
  132.  
  133.                         return Requests.indexRequest()
  134.                                 .index("flink_secure_log")
  135.                                 .type("_doc")
  136.                                 .source(map);
  137.                     }
  138.  
  139.                     @Override
  140.                     public void process(SecureLog secureLog, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
  141.                         requestIndexer.add(createIndexRequest(secureLog));
  142.                     }
  143.                 }
  144.         );
  145.         // 设置批量写数据的缓冲区大小
  146.         esSinkBuilder.setBulkFlushMaxActions(1);
  147.  
  148.         // provide a RestClientFactory for custom configuration on the internally created REST client
  149.         Header[] defaultHeaders = new Header[]{new BasicHeader("Authorization", HttpUtils.getHeader())};
  150.         esSinkBuilder.setRestClientFactory(
  151.                 restClientBuilder -> {
  152.                     restClientBuilder.setDefaultHeaders(defaultHeaders);
  153.                 }
  154.         );
  155.  
  156.         secureLogDataStreamWithWaterMask.print();
  157.         secureLogDataStreamWithWaterMask.addSink(esSinkBuilder.build());
  158.  
  159.         env.execute("Flink add data source");
  160.     }
  161.  
  162.  
  163.  
  164.  
  165. }

3、连接elasticsearch时，如果设置了用户名活密码，需要再http请求的header中加入Authorization授权信息

  1. package com.felix.utils;
  2.  
  3. import org.apache.commons.codec.binary.Base64;
  4.  
  5. import java.nio.charset.Charset;
  6.  
  7. public class HttpUtils {
  8.  
  9.     private static final String APP_KEY = "elastic"; //es用户名
  10.     private static final String SECRET_KEY = "123456";//es访问密码
  11.  
  12.     /**
  13.      * 构造Basic Auth认证头信息
  14.      *
  15.      * @return
  16.      */
  17.     public static String getHeader() {
  18.         String auth = APP_KEY + ":" + SECRET_KEY;
  19.         byte[] encodedAuth = Base64.encodeBase64(auth.getBytes(Charset.forName("US-ASCII")));
  20.         String authHeader = "Basic " + new String(encodedAuth);
  21.         return authHeader;
  22.     }
  23.  
  24.  
  25. }
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小丑西瓜9/article/detail/606336
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号