我可以在终端上运行这个例子。我的终端命令是：

bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 examples/src/main/python/sql/streaming/structured_kafka_wordcount.py localhost:9092 subscribe test

现在我想在juypter python笔记本中运行它。我试着遵循这个（我可以运行链接中的代码）。但在我看来，失败了。以下是我的代码：

import os
os.environ['PYSPARK_SUBMIT_ARGS'] = "--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 pyspark-shell"

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split

bootstrapServers = "localhost:9092"
subscribeType = "subscribe"
topics = "test"

spark = SparkSession\
    .builder\
    .appName("StructuredKafkaWordCount")\
    .getOrCreate()

# Create DataSet representing the stream of input lines from kafka

lines = spark\
    .readStream\
    .format("kafka")\
    .option("kafka.bootstrap.servers", bootstrapServers)\
    .option(subscribeType, topics)\
    .load()\
    .selectExpr("CAST(value AS STRING)")

# Split the lines into words

words = lines.select(
    # explode turns each item in an array into a separate row
    explode(
        split(lines.value, ' ')
    ).alias('word')
)

# Generate running word count

wordCounts = words.groupBy('word').count()

# Start running the query that prints the running counts to the console

query = wordCounts\
    .writeStream\
    .outputMode('complete')\
    .format('console')\
    .start()

query.awaitTermination()

错误消息是：

Py4JJavaError                             Traceback (most recent call last)
<ipython-input-1-0344129c7d54> in <module>()
     14 
     15 # Create DataSet representing the stream of input lines from kafka
---> 16 lines = spark    .readStream    .format("kafka")    .option("kafka.bootstrap.servers", bootstrapServers)    .option(subscribeType, topics)    .load()    .selectExpr("CAST(value AS STRING)")
     ...

Py4JJavaError: An error occurred while calling o31.load.
: java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/StreamWriteSupport
    at java.base/java.lang.ClassLoader.defineClass1(Native Method)
    ...

有什么想法吗？谢谢！

更新：

我试着跟着答案走，但还是出错了。以下是我的程序。我查了一下有两个 kernel.json ，他们是

~/anaconda3/pkgs/ipykernel-4.6.1-py36h3208c25_0/share/jupyter/kernels/python3/kernel.json
~/anaconda3/share/jupyter/kernels/python3/kernel.json

然后我用以下内容更新了它们：

{
    "display_name": "PySpark",
    "language": "python",
    "argv": [ "</usr>/anaconda3/bin/python", "-m", "ipykernel", "-f", "  {connection_file}" ],
    "env": {
        "SPARK_HOME": "</usr>/projects/spark-2.3.0",
        "PYSPARK_PYTHON": "</usr>/anaconda3/bin/python",
        "PYTHONPATH": "</usr>/projects/spark-2.3.0/spark/python/:</usr>/projects/spark-2.3.0/spark/python/lib/py4j-0.10.6-src.zip",
        "PYTHONSTARTUP": "</usr>/projects/spark-2.3.0/python/pyspark/shell.py",
        "PYSPARK_SUBMIT_ARGS":  "--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 pyspark-shell"
    }
}

然后我得到如下错误：

WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.hadoop.security.authentication.util.KerberosUtil (file:/<usr>/projects/spark-2.3.0/assembly/target/scala-2.11/jars/hadoop-auth-2.6.5.jar) to method sun.security.krb5.Config.getInstance()
WARNING: Please consider reporting this to the maintainers of org.apache.hadoop.security.authentication.util.KerberosUtil