使用flinktable api 不能连接到elasticSearch

在使用flink sink 数据到es时发现程序不能连上dev es服务器，但是可以用postman连接到es,同时相同的程序如果sink到本地的es，数据可以同步到es。因此排除程序的问题。本地环境和dev环境的区别是本地es使用http,dev使用https,不知道如何才能连接上dev环境的es

                "       'connector' = 'elasticsearch-7',\n" +
                "       'hosts' = 'https://xx.xx.xx.xx:9200',\n" +
                "       'username' = 'elastic',\n" +
                "       'password' = 'xxxxxxxxx',\n" +
                "       'index' = 'alarmxxxx'\n" +
                ")")

展开

收起

sqbbpyvjgm5vw 2023-06-06 10:29:10 160 0

5 条回答

写回答

取消提交回答

wljslmz

公众号：网络技术联盟站，InfoQ签约作者，阿里云社区签约作者，华为云云享专家，BOSS直聘创作王者，腾讯课堂创作领航员，博客+论坛：https://www.wljslmz.cn，工程师导航：https://www.wljslmz.com

可能是由于证书问题导致的。HTTPS 协议下需要进行 SSL/TLS 加密和认证，需要使用 ES 服务器的证书来建立安全连接。

2023-06-06 17:12:05

赞同展开评论打赏
vohelon
从您的描述来看，问题可能在于您的 Flink 程序无法连接到 dev 环境的 Elasticsearch（ES）服务器。

由于本地的 ES 使用 HTTP，而 dev 环境使用 HTTPS，这可能是导致连接失败的原因之一。

为了解决这个问题，您需要检查以下几个方面：
1. 确认 dev 环境的 ES 服务器是否已启动并正在运行。
2. 确认您的程序是否已正确配置连接 dev 环境的 ES 服务器。如果您使用的是 HTTPS，您需要在代码中使用相应的证书或信任证书。
3. 确认防火墙是否已正确配置，以允许 Flink 程序连接到 dev 环境的 ES 服务器。如果您使用的是 HTTPS，您需要打开 9200 端口以允许 HTTPS 流量通过。
4. 确认 dev 环境的 ES 服务器是否已正确配置以允许外部连接。如果服务器是安全的，您需要在服务器上配置安全策略以允许 Flink 程序连接。
为了更好地诊断问题，您可以尝试使用 curl 命令或其他 HTTP 客户端工具来测试您的 ES 服务器。如果您无法连接到 ES 服务器，请检查您的网络连接和配置，以确保它们正确无误。如果您可以连接到 ES 服务器，但 Flink 程序仍无法连接，请检查程序的配置并尝试重新启动它。
2023-06-06 11:17:18

赞同展开评论打赏
三掌柜666

十分耕耘，一定会有一分收获！
楼主你好，如果要使用HTTPS协议连接到ES，请确保在Flink源代码中提供了正确的证书文件。这是因为ES要求客户端证书验证，才能建立安全连接。

首先，需要确认ES服务器上是否启用了HTTPS。其次，需要将ES证书导出并添加到Flink的证书库中。

以下是可以尝试的步骤：
1. 导出服务器证书
使用浏览器或OpenSSL命令行工具来获取服务器证书。例如，对于使用Google Chrome浏览器访问ES的情况，您可以在Chrome浏览器中打开ES页面，单击锁图标，然后选择“证书信息”选项。接下来，在“详细信息”选项卡中，选择“导出”并保存证书文件。
1. 将证书添加到Flink的证书库中
将证书文件导入到Flink的证书库中。在Flink的配置文件中设置以下属性：
```
ssl.truststore.path=/path/to/truststore.jks
ssl.truststore.password=password
```
请注意替换路径和密码值。此外，也可以在代码中使用以下方法设置证书库：
```
env.getConfig().setString("security.ssl.truststore.path", "/path/to/truststore.jks");
env.getConfig().setString("security.ssl.truststore.password", "password");
```
1. 在sink的配置中启用HTTPS
在配置中启用HTTPS协议：
```
elasticsearchClientConfigBuilder.setProtocol(HttpHost.DEFAULT_SCHEME_NAME);
```
可以使用以下方法设置端口和主机名：
```
elasticsearchClientConfigBuilder.setHost("hostname", 443);
```
请注意替换实际的主机名和端口值。

以上步骤应该能够使您连接到ES服务器上的HTTPS端口，并将数据传输到ES中。
2023-06-06 11:00:27

赞同展开评论打赏
算精通

北京阿里云ACE会长
问题可能是由于您在使用Flink连接到Dev环境的Elasticsearch时，使用了HTTPS协议而不是HTTP协议，导致连接失败。您可以尝试以下步骤来解决这个问题：
1. 确定您的Dev环境的Elasticsearch是否在9200端口上监听HTTPS请求。如果不是，请确定正确的端口号。
2. 确认您的Dev环境中的Elasticsearch证书是否正确配置。如果证书未正确配置，则可能会导致连接失败。
3. 尝试在Flink中使用HTTPS协议连接到Dev环境的Elasticsearch。您可以尝试使用以下代码片段：
```
final ElasticsearchSink.Builder<T> esSinkBuilder = new ElasticsearchSink.Builder<>(
    httpsHosts,
    new ElasticsearchSinkFunction<T>() {...}
);
```
其中，httpsHosts是一个List<HttpHost>类型的变量，其中包含Elasticsearch服务器的HTTPS主机和端口信息。
1. 如果上述步骤都没有解决问题，您可以尝试使用其他方法连接到Dev环境的Elasticsearch，例如使用Java REST客户端API进行连接。这将需要更多的代码和配置，但可能能够解决您的问题。
2023-06-06 10:51:05

赞同展开评论打赏
魏红斌

天下风云出我辈，一入江湖岁月催，皇图霸业谈笑中，不胜人生一场醉。

可能需要在启动Flink程序时，通过-D参数指定es服务器的https连接方式。例如，可以使用以下命令启动Flink程序： flink run -c org.apache.flink.client.cli.CliFrontend ./flink-1.12-dist/bin/flink run -m yarn-cluster -yn 2 -ys 2 -yjm 1024 -ytm 1024 -c com.example.AlarmStreamingJob /path/to/alarm_streaming_job.jar /path/to/es_connector.properties 其中，-c参数指定启动的命令行参数，-m yarn-cluster表示使用yarn作为集群管理器，-yn 2表示使用2个YARN节点，-ys 2表示使用2个YARN节点，-yjm 1024表示使用1024个JobManager节点，-ytm 1024表示使用1024个TaskManager节点，-c com.example.AlarmStreamingJob表示使用com.example.AlarmStreamingJob作为Flink程序的名称，/path/to/alarm_streaming_job.jar表示Flink程序的jar包路径，/path/to/es_connector.properties表示es_connector.properties文件的路径。需要根据实际情况进行修改。

2023-06-06 10:40:08

赞同展开评论打赏