FlinkServer对接Elasticsearch

操作场景

本章节介绍通过使用FlinkServer写FlinkSQL对接Elasticsearch，Flink支持1.12.2及以后版本，Elasticsearch支持7.10.2及以后版本，详细如下：

• 安全模式的Flink集群支持对接安全模式和非安全模式的Elasticsearch集群。
  

  安全模式的Flink集群对接非安全模式的Elasticsearch集群时需设置如下参数：

  1. 登录Manager，选择“集群 > 服务 > Flink > 配置 > 全部配置”，搜索参数“es.security.indication”，并将FlinkResource角色和FlinkServer角色下该参数的值配置为“false” 。
  2. 重启Flink服务，在“概览”页签，选择“更多 > 重启服务”等待Flink服务重启成功。
• 非安全模式的Flink集群支持对接非安全模式的Elasticsearch集群。

本示例以安全模式FlinkServer对接本集群安全模式Elasticsearch为例。

前提条件

• 集群已安装Flink、Yarn、ZooKeeper、HDFS、Kafka和Elasticsearch服务。
• 参考创建FlinkServer角色创建一个具有FlinkServer管理员权限的用户用于访问Flink WebUI，如：flink_admin。并为该用户添加Elasticsearch用户组。若对接非本集群的Elasticsearch，请确保集群之间互信，并在待对接的集群内也创建具有相同权限的同名用户，并在Ranger UI中为该用户添加Elasticsearch的相关权限，可参考添加Elasticsearch的Ranger访问权限策略。
• 需确保Elasticsearch所在集群不存在“Elasticsearch服务存在red状态的索引”告警。

操作步骤

1. 使用flink_admin登录Manager，选择“集群 > 服务 > Flink”，在“Flink WebUI”右侧，单击链接，访问Flink的WebUI。
2. 参考新建作业，新建Flink SQL流作业，参考如下内容在作业开发界面进行作业开发，配置完成后启动作业。

   需勾选“基础参数”中的“开启CheckPoint”，“时间间隔（ms）”可设置为“60000”，“模式”可使用默认值。

   --创建KafkaSource表
   CREATE TABLE kafka(uuid STRING, name STRING, age INT) WITH (
     'connector' = 'kafka',
     'topic' = 'kafka2ES',
     'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
     'properties.group.id' = 'testgroup2',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'csv',
     'properties.sasl.kerberos.service.name' = 'kafka',
     'properties.security.protocol' = 'SASL_PLAINTEXT',
     'properties.kerberos.domain.name' = 'hadoop.系统域名'
   );
   --创建ESSink表
   CREATE TABLE es (uuid STRING, name STRING, age INT, 
   PRIMARY KEY (uuid) NOT ENFORCED
   ) WITH (
     'connector' = 'elasticsearch-7',
     'hosts' =  'https://10.10.10.10:24100;https://10.10.10.11:24100',
     'index' = 'test_index',
     'sink.bulk-flush.max-actions' = '1'
   );
   --将Kafka的数据写入ES
   insert into
     es
   select
     *
   from
     kafka;

   

   • Kafka Broker实例IP地址及端口号说明：
     • 服务的实例IP地址可通过登录FusionInsight Manager后，单击“集群 > 服务 > Kafka > 实例”，在实例列表页面中查询。
     • 集群已启用Kerberos认证（安全模式）时Broker端口为“sasl.port”参数的值，默认为“21007”。
     • 集群未启用Kerberos认证（普通模式）时Broker端口为“port”的值，默认为“9092”。如果配置端口号为9092，则需要配置“allow.everyone.if.no.acl.found”参数为true，具体操作如下：

       登录FusionInsight Manager系统，选择“集群 > 服务 > Kafka > 配置 > 全部配置”，搜索“allow.everyone.if.no.acl.found”配置，修改参数值为true，保存配置即可。

   • 系统域名：可登录FusionInsight Manager，选择“系统 > 权限 > 域和互信”，查看“本端域”参数，即为当前系统域名。
   • 10.10.10.10：Elasticsearch的EsNode的业务IP。登录FusionInsight Manager，选择“集群 > 服务 > Elasticsearch > 实例”，可查看所有Elasticsearch的EsNode的业务IP地址。
   • 24100：Elasticsearch默认端口号，可登录FusionInsight Manager系统，选择“集群 > 服务 > Elasticsearch > 配置 > 全部配置”，搜索“SERVER_PORT”查看EsNode的端口。

3. 查看作业管理界面，作业状态为“运行中”。
4. 参考管理Kafka主题中的消息，查看Topic并向Kafka中写入数据。

   ./kafka-topics.sh --list --zookeeper ZooKeeper的quorumpeer实例业务IP:ZooKeeper客户端端口号/kafka

   sh kafka-console-producer.sh --broker-list Kafka角色实例所在节点的IP地址:Kafka端口号 --topic 主题名称 --producer.config 客户端目录/Kafka/kafka/config/producer.properties

   例如本示例使用主题名称为kafka2ES：

   sh kafka-console-producer.sh --broker-list Kafka角色实例所在节点的IP地址:Kafka端口号 --topic kafka2ES --producer.config /opt/client/Kafka/kafka/config/producer.properties

   输入消息内容：

   3,zhangsan,18
   4,wangwu,19
   8,zhaosi,20

   输入完成后按回车发送消息。

   

   • ZooKeeper的quorumpeer实例业务IP：

     ZooKeeper服务所有quorumpeer实例业务IP。登录FusionInsight Manager，选择“集群 > 服务 > ZooKeeper > 实例”，可查看所有quorumpeer实例所在主机业务IP地址。

   • ZooKeeper客户端端口号：

     登录FusionInsight Manager，选择“集群 > 服务 > ZooKeeper”，在“配置”页签查看“clientPort”的值。

5. 参考Linux下curl命令的使用使用curl命令获取写入的数据，数据输出正常则表明成功对接Elasticsearch，如下图所示。

   如：curl -XGET --tlsv1.2 --negotiate -k -u : 'https://10.10.10.10:24100/test_index/_search?pretty'

   

WITH主要参数说明