也许外面有人可以帮助我。 我正在尝试使用 PySpark 从 ES 读取数据。我的 Jupyter Notebook 代码非常简单:
import pyspark
conf = pyspark.SparkConf().setAppName('Test').setMaster('spark://spark-master:7077')
sc = pyspark.SparkContext(conf=conf)
es_rdd = sc.newAPIHadoopRDD(
inputFormatClass="org.elasticsearch.hadoop.mr.EsInputFormat",
keyClass="org.apache.hadoop.io.NullWritable",
valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
conf={
"es.resource": "some-log/doc",
"es.nodes": "192.168.1.25",
"es.port": "9200"
})
我在运行 NB 的主机上安装了 Spark 和 Jupyter Notebook。 spark-defaults.conf 文件通过以下方式加载“elasticsearch-hadoop-6.4.0.jar”:spark.jars/opt/maya/es-hadoop/elasticsearch-hadoop-6.4.0.jar
我可以连接到 ES 实例并使用其他工具(如 elasticsearch-py)从中读取，Test 应用程序显示在 Spark Master UI 中。但是，当我执行上面的代码时，我不断收到此错误:
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-5-c990f37c388b> in <module>
6 "es.resource": "logs-dfir-winevent-security-*/doc",
7 "es.nodes": "192.168.248.131",
----> 8 "es.port": "9200"
9 })
10 #es_rdd.first()
/opt/anaconda/lib/python3.6/site-packages/pyspark/context.py in newAPIHadoopRDD(self, inputFormatClass, keyClass, valueClass, keyConverter, valueConverter, conf, batchSize)
715 jrdd = self._jvm.PythonRDD.newAPIHadoopRDD(self._jsc, inputFormatClass, keyClass,
716 valueClass, keyConverter, valueConverter,
--> 717 jconf, batchSize)
718 return RDD(jrdd, self)
719
/opt/anaconda/lib/python3.6/site-packages/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:
/opt/anaconda/lib/python3.6/site-packages/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.newAPIHadoopRDD.
: java.lang.ClassNotFoundException: org.elasticsearch.hadoop.mr.LinkedMapWritable
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at org.apache.spark.util.Utils$.classForName(Utils.scala:238)
at org.apache.spark.api.python.PythonRDD$.newAPIHadoopRDDFromClassNames(PythonRDD.scala:302)
at org.apache.spark.api.python.PythonRDD$.newAPIHadoopRDD(PythonRDD.scala:286)
at org.apache.spark.api.python.PythonRDD.newAPIHadoopRDD(PythonRDD.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 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)
我已经搜索过，但看不出错误出在代码本身，我感觉这个问题与运行 Jupyter Notebook 的主机中错误的 Spark 配置有关。任何见解将不胜感激!
最佳答案
请引用这个问题:pyspark: ship jar dependency with spark-submit
您需要做的是将依赖项的 jar 与配置一起传递。如果您使用的是 Jupyter notebook，则可以通过 SparkConf() 添加它，例如:
conf = SparkConf().option('spark.driver.extraClassPath', 'full/path/to/jar')
只需将您的代码更改为:
conf = pyspark.SparkConf().setAppName('Test').setMaster('spark://spark-master:7077').option('spark.driver.extraClassPath', 'full/path/to/jar')
关于apache-spark - 无法使用 PySpark 从 Elasticsearch 读取，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/52659109/