pyspark-Kafka整合：缺失自由

Question

为了开始Kafka的项目，我在这里按照Databricks的指示行事：

结构化流媒体Kafka集成指南（Kafka broker版本0.10.0或更高版本）

代码：

# coding: utf-8
import sys
import os,time
sys.path.append("/usr/local/lib/python2.7/dist-packages")
from pyspark.sql import SparkSession,Row
from pyspark import SparkContext,SQLContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from pyspark.sql.types import *
import pyspark.sql.functions
import json

spark = SparkSession.builder.appName("Kakfa-test").getOrCreate()
spark.sparkContext.setLogLevel('WARN')


trainingSchema = StructType([
  StructField("code",StringType(),True),
  StructField("ean",StringType(),True),
  StructField("name",StringType(),True),
  StructField("description",StringType(),True),
  StructField("category",StringType(),True),
  StructField("attributes",StringType(),True)
])
trainingDF = spark.createDataFrame(sc.emptyRDD(),trainingSchema)

broker, topic = 
['kafka.partner.stg.some.domain:9092','hybris.products']

df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", 
"kafka.partner.stg.some.domain:9092") \
.option("subscribe", "hybris.products") \
.option("startingOffsets", "earliest") \
.load()

我的Hadoop版本是2.6，Spark版本是2.3.0

带有spack-提交的命令行是：

spark提交——jars-jars/spark-sql-kafka-0-10_2.11-2.3.0。jar kafka-test-002。py

错误消息：

Py4JJavaError：在elation.scala:33010调用ource.scala:94时发生错误：org.apache.spark.sql.execution.streaming.NoClassDefFoundError：org/apache/kafka/公共/序列化/ByteArrayDeserializer。KafkaSourceProvider$.（KafkaSourcePorg.apache.spark.sql.streaming.）在eader.load010。KafkaSourceProvider$.（KafkaSourcePeader.scala:170）在sun.reflect.010。KafkaSourcePmpl.invokeStreamOptions（KafkaSourceProvider.scala:360）在org.apache.spark.sql.kafka010。KafkaSourceProvider.sourceSchema（KafkaSourceProvider.scala:64）在org.apache.spark.sql.execution.datasources.DataSource.sourceSchema（DataSource.scala:231）在org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute（DataSource.scala:94）在org.apache.spark.sql.execution.datasources.DataSource.sourceInfo（DataSo48.load.）在java.lang.StreamingRelation$. Application（StreamingRorg.apache.spark.sql.kafka）在rovider.scala:413DataStreamRorg.apache.spark.sql.kafka（DataStreamRrovider.scala）在org.apache.spark.sql.kafkaNativeMethodAccessorIrovider.validate0（本机方法URLClassLoader. findClass（URLClassLoader. java：381）在java. lang。ClassLoader. loadClass（ClassLoader. java：424）在java. lang。ClassLoader. loadClass（ClassLoader. java：357）

正如你可以在我上面提到的网站上看到的，我正在导入的jar文件是完全相同的文件。所以，我不知道为什么会这样。也许还有一个模块没提到？我真的迷路了

Answer 1

提到的JAR不包括kafka客户端的所有依赖项。你更应该使用--packages org。阿帕奇。spark:spark-sql-kafka-0-10_2.11:2.3.0（如部署部分的文档所述：https://spark.apache.org/docs/2.3.0/structured-streaming-kafka-integration.html#deploying)