Apache Flink v1.14无OpenSearch连接器，如何实现结果写入？

由于OpenSearch是Elasticsearch的衍生项目，二者API层面高度兼容，你可以直接复用Flink的Elasticsearch连接器实现写入，以下是具体可行方案：

1. 用Elasticsearch 7.x连接器适配OpenSearch #

Flink 1.14的Elasticsearch 7连接器可直接对接OpenSearch，仅需调整连接配置：

• 引入依赖（Maven项目示例）：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch7_2.12</artifactId>
    <version>1.14.6</version>
</dependency>

• 核心代码实现：

List<HttpHost> hosts = new ArrayList<>();
hosts.add(new HttpHost("你的OpenSearch节点地址", 9200, "http"));

ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
    hosts,
    new ElasticsearchSinkFunction<String>() {
        public IndexRequest createIndexRequest(String element) {
            Map<String, String> json = new HashMap<>();
            json.put("data", element);

            return Requests.indexRequest()
                .index("目标索引名")
                .source(json, XContentType.JSON);
        }

        @Override
        public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
            indexer.add(createIndexRequest(element));
        }
    }
);

// 配置批量写入规则
esSinkBuilder.setBulkFlushMaxActions(1000);
esSinkBuilder.setBulkFlushInterval(1000);

// 将sink添加到Flink数据流
dataStream.addSink(esSinkBuilder.build());

• 安全认证配置（如果OpenSearch开启了账号密码验证）：

CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY,
    new UsernamePasswordCredentials("用户名", "密码"));

RestClientBuilder builder = RestClient.builder(hosts.toArray(new HttpHost[0]))
    .setHttpClientConfigCallback(httpClientBuilder -> 
        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));

esSinkBuilder.setRestClientFactory(restClientBuilder -> {
    restClientBuilder.setHttpClientConfigCallback(builder.getHttpClientConfigCallback());
});

2. 自定义Sink（备选方案） #

如果上述兼容方案遇到特殊适配问题，也可以基于OpenSearch官方Java客户端自定义Flink Sink，但这种方式需要自行实现批量写入、故障重试、状态管理等逻辑，开发成本较高，非必要不推荐。

关键注意事项 #

• 确认OpenSearch版本与Elasticsearch 7.x的API兼容性：OpenSearch 1.x完全兼容ES 7.x；OpenSearch 2.x部分API有调整，需针对性修改参数适配。
• 测试写入后的索引映射是否符合预期，避免字段类型不匹配导致写入失败。

内容的提问来源于stack exchange，提问作者Abhinav