Pyspark Kafka ReadStream

Question

我正在使用以下代码读取KAFKA主题的数据。

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.3,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 pyspark-shell'

df = spark.readStream.format("kafka").option("kafka.bootstrap.server", kafkaServer).option("subscribe", topic_name_read).option("includeHeaders", "true").option("startingOffsets", "earliest").load()

但这引发了以下错误。

文件“ c：\ spark \ spark \ spark-3.1.3 bin-hadoop2.7 \ python \ python \ pyspark \ sql \ sql \ sql \ readwriter.py”，第210行，in LoadRetReturn self._df（self._jreader.load（））文件“ c：c：\ spark \ spark-3.1.3 bin-hadoop2.7 \ python \ python \ lib \ py4j-0.10.9-src.zip \ zip \ py4j \ java_gateway.py”，第1304行\ spark-3.1.3 bin-hadoop2.7 \ python \ python \ pyspark \ sql \ sql \ utils.py”，第111行，在decoreturn f（*a，** kw）文件中-bin-hadoop2.7 \ python \ lib \ py4j-0.10.9-src.zip \ py4j \ progent.py.py”，第326行，在get_return_valuepy4j.protot.prot.py4jjavaerror中lang.noclassdeffounderror：org/apache/kafka/common/serialization/bytearrayserializerait org.apache.spark.sql.kafka010.kafkasourceprovider $。 kafkasourceprovider $。在org.apache .spark.sql.kafka010.kafkasourceprovider.createeralation（kafkasourceprovider.scala：127）at org.apache.spark.sql.sql.squt.sql.execution.datasources.datasources.datasource.datasource.datasource.resolce.resolverelation（datasOlce.scalce.scala：355） 。 spark.sql.dataframereader.load（dataframereader.scala：307）at org.apache.spark.sql.sql.dataframereader.load（dataframereader.scala：225）在sun.reflect.reflect.nativemethodaccesorimpl.invoke.invoke0（sun.nativemethodemethodemethodaccesorimple.invoke0（sun） nativemethodaccesorimpl.invoke（nativemethodaccesorimpl.java：62）在sun.reflect.delegatingmethodaccesorimpl.invoke（java.lang.lang.reflect.reflect.method.method.in.method.method.method.methode.mjrece.mjava：43） ETHODINVOKER.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)请访问py4j.commands.callcommand.execute（callcommand.java：79），py4j.gatewaywayconnection.run（gatewayconnection.java：238），java.lang.thread.thread.run（thread.java：750），引起：java.lang。 classNotFoundException：org.apache.kafka.common.serialization.bytearrayserializerAsat Java.net.urlclassloader.findclass（urlclassloader.java：387） 。

​功能。

df1 = df.withColumn("items", F.explode(F.col("items")))

attributeError：'function'对象没有属性'带有column'

谢谢

环境：

python -3.9.9

pyspark -3.1.3

kafka -python -2.0.2

spark -hadoop -3.1.3和2.7

< strong> jar使用

spark-streaming-kafka-0-10_2.12-3.1.3，

spark-sql-kafka-0-10_2.12-3.1.3

Answer 1

.load（）是正确的。

您的真正错误：

classNotFoundException：org.apache.kafka.common.serialization.bytearrayserializer

您需要在您的中包括kafka-clients依赖项 - packages

您仅在此处使用Spark-SQL-KAFKA，因此您应该删除另一个。