所以在我开始之前，我想说我知道这个问题可能看起来重复或重复，但那里可用的答案与我的要求无关。它们主要用于java和scala，但只有一点关于python。
所以我有一个在 gcp 上运行的 kerberos 安全的 kafka 集群。我通过使用 kafka-python 包创建生产者和消费者来检查两次正在运行，它运行良好。
但是当我尝试使用 pyspark 使用我的 Spark 应用程序连接到该集群时，它不起作用。我的火花应用程序如下所示：-
def application(topic, batchTime, appName, **kwargs):
import os
try:
os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars <full-path-to-spark>/jars/spark-streaming-kafka-0-8-assembly_2.11-2.4.4.jar pyspark-shell'
conf = SparkConf().setAppName(appName).setMaster('local[*]')
sc = SparkContext(conf=conf)
stream_context = StreamingContext(sparkContext=sc, batchDuration=batchTime)
kafka_stream = KafkaUtils.createDirectStream(ssc=stream_context, topics=[topic],
kafkaParams={"metadata.broker.list":"broker1:9092",
"ssl.context": 'context',
'sasl.plain.username': '****',
'sasl.plain.password': '*********',
'sasl.mechanism': 'PLAIN',
'security.protocol': "SASL_PLAINTEXT"})
lines = kafka_stream.map(lambda x: json.loads(x[1]))
final_obj = lines.map(lambda line: SparkHelper.get_app_type(line, line['app_type']))
final_obj.foreachRDD(handler)
final_obj.pprint()
当我运行它时，它显示错误，例如：-
19/11/26 19:11:59 WARN Utils: Your hostname, openstack-inspiron-3543 resolves to a loopback address: 127.0.1.1; using 10.10.0.25 instead (on interface wlp6s0)
19/11/26 19:11:59 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
19/11/26 19:11:59 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
19/11/26 19:12:01 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
19/11/26 19:12:01 WARN VerifiableProperties: Property sasl.mechanism is not valid
19/11/26 19:12:01 WARN VerifiableProperties: Property sasl.plain.password is not valid
19/11/26 19:12:01 WARN VerifiableProperties: Property sasl.plain.username is not valid
19/11/26 19:12:01 WARN VerifiableProperties: Property security.protocol is not valid
19/11/26 19:12:01 WARN VerifiableProperties: Property ssl.context is not valid
控制台卡在那里。它不会终止也不会运行任何内容。是的，这仅在我的本地 kafka 集群运行时发生。如果我关闭它并运行 spark 应用程序，它会显示"NoBrokerAvailable"。这不应该发生吧？当我尝试一起连接到不同的集群时，我的本地 kafka 集群是否正在运行并不重要。
我尝试安装一个不同的 spark-stream-kafka-assembly jar 文件，但它们都不起作用。我从 https://jar-download.com/?search_box=spark-streaming-kafka-assembly 下载了jar文件。
我正在使用的当前版本是...
火花：2.4.4， Hadoop： 2.7， api_version： 0.10
我正在使用融合平台 5.3.1 来运行 kafka 集群。
我不明白问题出在哪里。请指出我犯的任何错误，或者如果没有，我应该更改什么才能完成这项工作。提前感谢您的回复！！
您可以尝试 Pyspark 结构化流而不是 DSstream 方法，因为它可以选择传递 SSL 功能并轻松处理。请参考下面堆栈溢出链接中测试的示例代码。 找不到主题的领导者;java.lang.NullPointerException NullPointerException at org.apache.kafka.common.utils.Utils.formatAddress