我想用Pypark给Kafka写信。
我被困在零号舞台上：

[Stage 0:>                                                          (0 + 8) / 9]

然后我得到一个超时错误： org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms. 代码为：

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

from pyspark.sql.functions import *
from pyspark.sql import SparkSession
from pyspark.sql.types import *

def main():
    spark = SparkSession.builder.master("local").appName("Spark CSV Reader")
     .getOrCreate();

    dirpath =  os.path.abspath(sys.argv[1])
    os.chdir(dirpath)

    mySchema = StructType([
     StructField("id", IntegerType()),StructField("name", StringType()),\
     StructField("year", IntegerType()),StructField("rating", DoubleType()),\
     StructField("duration", IntegerType())   ])
    streamingDataFrame = spark.readStream.schema(mySchema)
     .csv('file://' + dirpath + "/" )

    streamingDataFrame.selectExpr("CAST(id AS STRING) AS key",
     "to_json(struct(*)) AS value").\
      writeStream.format("kafka").option("topic", "topicName")\
      .option("kafka.bootstrap.servers", "localhost:9092")\
      .option("checkpointLocation", "./chkpt").start()

我正在运行hdp2.6。