如何从 Zeppelin 连接到安全的 Kafka 集群（“无法构建 kafka 消费者”）？

Question

我正在尝试使用结构化流从 Kafka 代理读取一些数据，以将其显示在 Zeppelin 笔记中。我在 Heroku 上使用启用了 SSL 的 Spark 2.4.3、Scala 2.11、Python 2.7、Java 9 和 Kafka 2.2，但得到 StreamingQueryException: 'Failed to construction kafka consumer'。

我正在使用以下依赖项（在 Spark 解释器设置中设置）：

org.apache.spark:spark-streaming-kafka-0-10_2.11:2.4.3  
org.apache.spark:spark-streaming_2.11:2.4.3     
org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.3 

我尝试过旧版本和新版本，但这些版本应该与我正在使用的 Spark/Scala 版本相匹配。

我已经使用简单的 Python 生产者和消费者成功地从 Kafka 写入和读取。

我正在使用的代码：

%pyspark
from pyspark.sql.functions import from_json
from pyspark.sql.types import *
from pyspark.sql.functions import col, expr, when

schema = StructType().add("power", IntegerType()).add("colorR", IntegerType()).add("colorG",IntegerType()).add("colorB",IntegerType()).add("colorW",IntegerType())

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", brokers) \
  .option("kafka.security.protocol", "SSL") \
  .option("kafka.ssl.truststore.location", "/home/ubuntu/kafka/truststore.jks") \
  .option("kafka.ssl.keystore.location", "/home/ubuntu/kafka/keystore.jks") \
  .option("kafka.ssl.keystore.password", password) \
  .option("kafka.ssl.truststore.password", password) \
  .option("kafka.ssl.endpoint.identification.algorithm", "") \
  .option("startingOffsets", "earliest") \
  .option("subscribe", topic) \
  .load()

schema = ArrayType(
    StructType([StructField("power", IntegerType()), 
                StructField("colorR", IntegerType()),
                StructField("colorG", IntegerType()),
                StructField("colorB", IntegerType()),
                StructField("colorW", IntegerType())]))

readDF = df.select( \
  col("key").cast("string"),
  from_json(col("value").cast("string"), schema))

query = readDF.writeStream.format("console").start()
query.awaitTermination()

我得到的错误：

Fail to execute line 43: query.awaitTermination()
Traceback (most recent call last):
  File "/tmp/zeppelin_pyspark-2171412221151055324.py", line 380, in <module>
    exec(code, _zcUserQueryNameSpace)
  File "<stdin>", line 43, in <module>
  File "/home/ubuntu/spark/python/lib/pyspark.zip/pyspark/sql/streaming.py", line 103, in awaitTermination
    return self._jsq.awaitTermination()
  File "/home/ubuntu/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "/home/ubuntu/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 75, in deco
    raise StreamingQueryException(s.split(': ', 1)[1], stackTrace)
StreamingQueryException: u'Failed to construct kafka consumer\n=== Streaming Query ===\nIdentifier: [id = 2ee20c47-8293-469a-bc0b-ef71a1f118bc, runId = 72422290-090a-4b6d-bd66-088a5a534240]\nCurrent Committed Offsets: {}\nCurrent Available Offsets: {}\n\nCurrent State: ACTIVE\nThread State: RUNNABLE\n\nLogical Plan:\nProject [cast(key#7 as string) AS key#22, jsontostructs(ArrayType(StructType(StructField(power,IntegerType,true), StructField(colorR,IntegerType,true), StructField(colorG,IntegerType,true), StructField(colorB,IntegerType,true), StructField(colorW,IntegerType,true)),true), cast(value#8 as string), Some(Etc/UTC)) AS jsontostructs(CAST(value AS STRING))#21]\n+- StreamingExecutionRelation KafkaV2[Subscribe[tanana-44614.lightbulb]], [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13]\n'

当我使用读写而不是 readStream 和 writeStream 时，我没有收到任何错误，但是当我向 Kafka 发送一些数据时，控制台上什么也没有显示。

我还应该尝试什么？

Answer 1

看起来 Kafka 消费者无法访问 ~/kafka/truststore.jks，因此出现异常。将 ~ 替换为完全指定的路径（不带波浪号），问题应该会消失。