seatunnel 怎么配置Elasticsearch

Question

seatunnel 怎么配置Elasticsearch，需要携带证书，

===============================================================================


2024-05-06 15:54:39,748 ERROR [o.a.s.c.s.SeaTunnel           ] [main] - Fatal Error, 

2024-05-06 15:54:39,748 ERROR [o.a.s.c.s.SeaTunnel           ] [main] - Please submit bug report in https://github.com/apache/seatunnel/issues

2024-05-06 15:54:39,748 ERROR [o.a.s.c.s.SeaTunnel           ] [main] - Reason:SeaTunnel job executed failed 

2024-05-06 15:54:39,749 ERROR [o.a.s.c.s.SeaTunnel           ] [main] - Exception StackTrace:org.apache.seatunnel.core.starter.exception.CommandExecuteException: SeaTunnel job executed failed
    at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:199)
    at org.apache.seatunnel.core.starter.SeaTunnel.run(SeaTunnel.java:40)
    at org.apache.seatunnel.core.starter.seatunnel.SeaTunnelClient.main(SeaTunnelClient.java:34)
Caused by: org.apache.seatunnel.api.table.catalog.exception.CatalogException: ErrorCode:[API-03], ErrorDescription:[Catalog initialize failed] - Failed to open catalog Elasticsearch
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.catalog.ElasticSearchCatalog.open(ElasticSearchCatalog.java:85)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.sink.ElasticsearchSink.getSaveModeHandler(ElasticsearchSink.java:96)
    at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.handleSaveMode(MultipleTableJobConfigParser.java:664)
    at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.createSinkAction(MultipleTableJobConfigParser.java:655)
    at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parseSink(MultipleTableJobConfigParser.java:567)
    at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parse(MultipleTableJobConfigParser.java:193)
    at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.getLogicalDag(ClientJobExecutionEnvironment.java:88)
    at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.execute(ClientJobExecutionEnvironment.java:161)
    at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:146)
    ... 2 more
Caused by: java.lang.RuntimeException: java.io.IOException: Invalid keystore format
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient.lambda$getRestClientBuilder$2(EsRestClient.java:189)
    at org.elasticsearch.client.RestClientBuilder.createHttpClient(RestClientBuilder.java:213)
    at java.security.AccessController.doPrivileged(Native Method)
    at org.elasticsearch.client.RestClientBuilder.build(RestClientBuilder.java:189)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient.createInstance(EsRestClient.java:132)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient.createInstance(EsRestClient.java:99)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.catalog.ElasticSearchCatalog.open(ElasticSearchCatalog.java:76)
    ... 10 more
Caused by: java.io.IOException: Invalid keystore format
    at sun.security.provider.JavaKeyStore.engineLoad(JavaKeyStore.java:669)
    at sun.security.provider.JavaKeyStore$JKS.engineLoad(JavaKeyStore.java:57)
    at sun.security.provider.KeyStoreDelegator.engineLoad(KeyStoreDelegator.java:224)
    at sun.security.provider.JavaKeyStore$DualFormatJKS.engineLoad(JavaKeyStore.java:71)
    at java.security.KeyStore.load(KeyStore.java:1449)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.util.SSLUtils.createSSLContext(SSLUtils.java:86)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.util.SSLUtils.buildSSLContext(SSLUtils.java:59)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient.lambda$getRestClientBuilder$2(EsRestClient.java:172)
    ... 16 more
 
2024-05-06 15:54:39,749 ERROR [o.a.s.c.s.SeaTunnel           ] [main] - 
===============================================================================



Exception in thread "main" org.apache.seatunnel.core.starter.exception.CommandExecuteException: SeaTunnel job executed failed
    at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:199)
    at org.apache.seatunnel.core.starter.SeaTunnel.run(SeaTunnel.java:40)
    at org.apache.seatunnel.core.starter.seatunnel.SeaTunnelClient.main(SeaTunnelClient.java:34)
Caused by: org.apache.seatunnel.api.table.catalog.exception.CatalogException: ErrorCode:[API-03], ErrorDescription:[Catalog initialize failed] - Failed to open catalog Elasticsearch
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.catalog.ElasticSearchCatalog.open(ElasticSearchCatalog.java:85)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.sink.ElasticsearchSink.getSaveModeHandler(ElasticsearchSink.java:96)
    at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.handleSaveMode(MultipleTableJobConfigParser.java:664)
    at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.createSinkAction(MultipleTableJobConfigParser.java:655)
    at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parseSink(MultipleTableJobConfigParser.java:567)
    at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parse(MultipleTableJobConfigParser.java:193)
    at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.getLogicalDag(ClientJobExecutionEnvironment.java:88)
    at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.execute(ClientJobExecutionEnvironment.java:161)
    at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:146)
    ... 2 more
Caused by: java.lang.RuntimeException: java.io.IOException: Invalid keystore format
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient.lambda$getRestClientBuilder$2(EsRestClient.java:189)
    at org.elasticsearch.client.RestClientBuilder.createHttpClient(RestClientBuilder.java:213)
    at java.security.AccessController.doPrivileged(Native Method)
    at org.elasticsearch.client.RestClientBuilder.build(RestClientBuilder.java:189)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient.createInstance(EsRestClient.java:132)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient.createInstance(EsRestClient.java:99)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.catalog.ElasticSearchCatalog.open(ElasticSearchCatalog.java:76)
    ... 10 more
Caused by: java.io.IOException: Invalid keystore format
    at sun.security.provider.JavaKeyStore.engineLoad(JavaKeyStore.java:669)
    at sun.security.provider.JavaKeyStore$JKS.engineLoad(JavaKeyStore.java:57)
    at sun.security.provider.KeyStoreDelegator.engineLoad(KeyStoreDelegator.java:224)
    at sun.security.provider.JavaKeyStore$DualFormatJKS.engineLoad(JavaKeyStore.java:71)
    at java.security.KeyStore.load(KeyStore.java:1449)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.util.SSLUtils.createSSLContext(SSLUtils.java:86)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.util.SSLUtils.buildSSLContext(SSLUtils.java:59)
    at org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient.lambda$getRestClientBuilder$2(EsRestClient.java:172)
    ... 16 more

Answer 1

引自免费微信小程序：皆我百晓生

从错误信息来看，你在尝试使用 SeaTunnel (原名 Waterdrop) 将数据写入到 Elasticsearch，并且启用了 TLS 加密，但是遇到了 Invalid keystore format 的错误。这意味着 Seatunnel 在尝试加载你的证书文件时，发现证书文件的格式不被支持或者不正确。为了帮助你解决这个问题，我会提供一个详细的配置示例以及可能的解决方案。

配置示例

首先，确保你的 Elasticsearch 配置是正确的。在 Seatunnel 配置文件中，对于 Elasticsearch Sink，一个基本且携带证书的配置示例可能如下所示：

sink {
    Elasticsearch {
        source_table_name = "matter_result_data"
        hosts = ["10.3200"]
        # 注意：这里的路径应该是一个有效的PKCS12格式的证书文件路径
        tls_keystore_path = "/path/to/your/pkcs12_certificate.p12"
        # 对于PKCS12格式，密码是必需的
        tls_keystore_password = "your_keystore_password"
        username = "elastic"
        password = "your_elastic_password"
        index = "sc_zw_matter"
        schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
        data_save_mode = "APPEND_DATA"
        primary_keys = ["matterId"]
    }
}

解决方案

1. 检查证书格式：错误信息指出“Invalid keystore format”，这通常意味着你提供的证书文件不是一个有效的 Java KeyStore (JKS) 或 PKCS12 格式文件，而 Seatunnel 默认期望的是 PKCS12 格式。请确认你的 tls_keystore_path 指向的是一个 PKCS12 格式的证书文件（通常扩展名为 .p12 或 .pkcs12）。

2. 转换证书格式：如果你的证书是 JKS 格式而非 PKCS12，你需要将其转换为 PKCS12 格式。可以使用 keytool 命令行工具来完成这个操作：

   keytool -importkeystore -srckeystore /path/to/your/jks_file.jks -destkeystore /path/to/new/pkcs12_certificate.p12 -deststoretype PKCS12
   

   运行此命令时，它会提示你输入源密钥库的密码、目标密钥库的密码等信息。

3. 验证证书密码：确保在配置文件中指定的 tls_keystore_password 是正确的。密码是你在创建或转换证书时设置的那个。

4. 更新配置文件：根据上述步骤修改或确认配置后，更新你的 Seatunnel 配置文件，确保所有路径和密码都是正确的。

5. 重新启动 Seatunnel：完成上述更改后，重新启动 Seatunnel，看看问题是否得到解决。

如果问题仍然存在，可能需要进一步检查证书文件是否损坏，或者 Seatunnel 版本与证书兼容性问题。不过，以上步骤应该能解决大多数“Invalid keystore format”的情况。