问弹性搜索-无法初始化SSL -证书问题

EN

我试图通过PySpark从Elastic (版本:7.13.4)中获取数据。但是，我得到了这个错误。

org.elasticsearch.hadoop.EsHadoopIllegalStateException: Cannot initialize SSL - parseAlgParameters failed: ObjectIdentifier() -- data isn't an object ID (tag = 48)

我的代码：

import findspark
findspark.init()
from functools import reduce
import time, datetime, argparse,math,configparser,urllib3
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext,SparkSession,Window
from pyspark.sql.functions import split,lit,to_utc_timestamp,hour,dayofweek,round,lower,col,unix_timestamp,pandas_udf,PandasUDFType,mean,date_format,concat,when,first,last,count,sum,ceil,max,asc
from pyspark.sql.types import TimestampType,IntegerType,StringType
from datetime import timedelta
from elasticsearch import Elasticsearch
from elasticsearch.exceptions import NotFoundError
import pyspark.sql.functions as F
from pyspark.sql.functions import*

query = """{
  "query": {
    "match_all": {}
  }
}"""


df = spark.read.format("org.elasticsearch.spark.sql").option("es.nodes", "elasticsearch-svc.namespace") \
                                                         .option("es.port","9200") \
                                                         .option("es.read.metadata", "false") \
                                                         .option("es.mapping.date.rich", "false") \
                                                         .option("es.query", query) \
                                                         .option("es.net.http.auth.user", "elastic-username") \
                                                         .option("es.net.http.auth.pass", "elastic-password") \
                                                         .option("es.net.ssl.keystore.location","file:////my-storage/ssl_certificates/elastic-certificates.p12") \
                                                         .option("es.net.ssl.keystore.pass","mypassword") \
                                                         .option("es.net.ssl.keystore.type","PKCS12") \
                                                         .option("es.net.ssl.truststore.location","file:////my-storage/ssl_certificates/elastic-certificates.p12") \
                                                         .option("es.net.ssl.truststore.pass","mypassword") \
                                                         .option("es.net.ssl","true") \
                                                         .option("es.net.ssl.cert.allow.self.signed","true") \
                                                         .load("my_index")

完全错误：

Py4JJavaError:调用o673.load时出错。：：org.elasticsearch.hadoop.EsHadoopIllegalArgumentException:无法检测ES版本--通常情况下，如果无法访问网络/Elasticsearch集群，或者针对WAN/Cloud实例，而没有在org.elasticsearch.spark.sql.ElasticsearchRelation.cfg$lzycompute(DefaultSource.scala:225) at org.elasticsearch.spark.sql.ElasticsearchRelation.cfg(DefaultSource.scala:223) at org.elasticsearch.spark.sql.ElasticsearchRelation.lazySchema$lzycompute(DefaultSource上设置正确的“es.nodes.wan.only”，则会发生这种情况。org.elasticsearch.spark.sql.ElasticsearchRelation.lazySchema(DefaultSource.scala:229) at org.elasticsearch.spark.sql.ElasticsearchRelation$$anonfun$schema$1.apply(DefaultSource.scala:233) at org.elasticsearch.spark.sql.ElasticsearchRelation$$anonfun$schema$1.apply(DefaultSource.scala:233) at scala.Option.getOrElse(Option.scala:121) org.elasticsearch.spark.sql.ElasticsearchRelation.schema(DefaultSource.scala:233) at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:403) atorg.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223) at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211) at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:178) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) at py4j.Gateway.invoke(Gateway.java:282) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.GatewayConnection.run(GatewayConnection.java:238) at java.lang.Thread.run(Thread.java:748)由: org.elasticsearch.hadoop.EsHadoopIllegalStateException:无法初始化SSL - parseAlgParameters失败: ObjectIdentifier() --数据不是org.elasticsearch.hadoop.rest.commonshttp.SSLSocketFactory.getSSLContext(SSLSocketFactory.java:160) at org.elasticsearch.hadoop.rest.commonshttp.SSLSocketFactory.getSSLContext(SSLSocketFactory.java:160) at org.elasticsearch.hadoop.rest.commonshttp.SSLSocketFactory.createSocket(SSLSocketFactory.java:129) at org.apache.commons.httpclient.HttpConnection.open(HttpConnection.java:707) at org.apache.commons.httpclient.HttpMethodDirector.executeWithRetry(上的对象ID (tag = 48)org.apache.commons.httpclient.HttpMethodDirector.executeMethod(HttpMethodDirector.java:171) at org.apache.commons.httpclient.HttpClient.executeMethod(HttpClient.java:397) at org.apache.commons.httpclient.HttpClient.executeMethod(HttpClient.java:323) at org.elasticsearch.hadoop.rest.commonshttp.CommonsHttpTransport.doExecute(CommonsHttpTransport.java:685) at org.elasticsearch.hadoop.rest.commonshttp.CommonsHttpTransport.execute(CommonsHttpTransport.java:664) at org.elasticsearch.hadoop.rest.NetworkClient.execute(NetworkClient.org.elasticsearch.hadoop.rest.RestClient.execute(RestClient.java:432) at org.elasticsearch.hadoop.rest.RestClient.execute(RestClient.java:428) at org.elasticsearch.hadoop.rest.RestClient.execute(RestClient.java:388) at org.elasticsearch.hadoop.rest.RestClient.execute(RestClient.java:392) at org.elasticsearch.hadoop.rest.RestClient.get(RestClient.java:168) at org.elasticsearch.hadoop.rest.RestClient.mainInfo(RestClient.java:745) at org.elasticsearch.hadoop.rest.InitializationUtils.discoverClusterInfo(InitializationUtils.java:330) . 23更多的原因是: java.io.IOException: parseAlgParameters失败: ObjectIdentifier() --在sun.security.pkcs12.PKCS12KeyStore.engineLoad(PKCS12KeyStore.java:2018) at java.security.KeyStore.load(KeyStore.java:1445) at org.elasticsearch.hadoop.rest.commonshttp.SSLSocketFactory.loadKeyStore(SSLSocketFactory上，数据不是对象ID (tag = 48)在org.elasticsearch.hadoop.rest.commonshttp.SSLSocketFactory.loadKeyManagers(SSLSocketFactory.java:215) at org.elasticsearch.hadoop.rest.commonshttp.SSLSocketFactory.createSSLContext(SSLSocketFactory.java:173) .还有40多个原因是: java.io.IOException: ObjectIdentifier() --数据不是一个对象ID (tag = 48)，而是sun.security.util.ObjectIdentifier.(ObjectIdentifier.java:257) at sun.security.util.DerInputStream.getOID(DerInputStream.java:314) at com.sun。crypto.provider.PBES2Parameters.engineInit(PBES2Parameters.java:267) at java.security.AlgorithmParameters.init(AlgorithmParameters.java:293) at sun.security.pkcs12.PKCS12KeyStore.parseAlgParameters(PKCS12KeyStore.java:812) . 45

我认为keystore类型可能是问题所在，并试图将.p12文件转换为.jks文件，而且它也不起作用。

错误：

keytool error: java.io.IOException: parseAlgParameters failed: ObjectIdentifier() -- data isn't an object ID (tag = 48)