Flink 输出到 Elasticsearch 报错 ElasticsearchStatusException 如何解决？

2023-08-02 13:04:11

在 Flink 中解决 ElasticsearchStatusException：missing authentication credentials for REST request 错误

错误

当您在 Flink 中使用 Elasticsearch 输出时，如果 Elasticsearch 启用了账号密码模式，并且您没有提供正确的账号和密码，就会抛出 ElasticsearchStatusException：missing authentication credentials for REST request 错误。这是因为 Elasticsearch 需要验证您的身份，才能允许您访问它的数据。

解决方法

要解决此错误，需要在 Flink 中配置正确的 Elasticsearch 账号和密码。具体步骤如下：

步骤 1：获取 Elasticsearch 账号和密码

从 Elasticsearch 管理员处获取账号和密码。
自行创建账号和密码。

步骤 2：在 Flink 配置文件中添加 Elasticsearch 账号和密码

在 Flink 配置文件中找到以 "elasticsearch" 开头的配置项，并在其中添加以下内容：

username: "您的 Elasticsearch 账号"
password: "您的 Elasticsearch 密码"

步骤 3：重新启动 Flink 作业

配置好 Elasticsearch 账号和密码后，重新启动 Flink 作业，以便新配置生效。

代码示例

// 配置 Elasticsearch 连接
ElasticsearchUpsertBulkWriter.Builder builder = ElasticsearchUpsertBulkWriter.newBuilder();
builder.setBulkFlushMaxActions(100); // 批量操作的最大动作数
builder.setBulkFlushMaxSizeMb(10); // 批量操作的最大大小（MB）
builder.setFlushIntervalMillis(1000); // 刷新间隔（毫秒）
builder.setElasticsearchHttpClientConfig(elasticsearchHttpClientConfig); // Elasticsearch HTTP 客户端配置
builder.setUsername("您的 Elasticsearch 账号");
builder.setPassword("您的 Elasticsearch 密码");

// 创建 Elasticsearch 输出
ElasticsearchUpsertBulkWriter writer = builder.build();