如何将 pyspark 流验证为安全的 sasl 普通卡夫卡？

Question

我正在尝试从 spark 流式传输到安全的 kafka 代理（使用 SASL PLAINTEXT 机制）。 我使用的是使用 spark-streaming-kafka-0-8 版本的 Pyspark，我使用的是 kafka 代理版本 0.10.2。

我遇到的问题是，当 pyspark 流式传输到 kafka 时不会在输出中返回任何内容。

我的问题是：

• 如何正确地从 spark 流式验证到安全的 sasl 普通 kafka？
• 或者 pyspark 是否支持对安全的 sasl 普通 kafka 进行身份验证？我很好奇 pyspark 仅受 spark-streaming-kafka-0-8 支持。因为在该版本中，对 spark-streaming-kafka-0-8 的初始支持是 kafka 代理版本 0.8.2，因为版本 0.9 支持 kafka 身份验证。

补充：

• 我可以使用控制台工具发布和订阅，因为我已为我的用户设置了我想要使用的主题的 acl。

这里有可能解释问题的细节。

在 server.properties 中为 kafka 代理启用 sasl plain 的附加参数

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
super.users=User:admin

这也是我的 kafka_jaas_server.conf

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin1!"
    user_admin="admin1!"
    user_aldys="admin1!";
};

在启动 kafka 代理时，我还包括了我的 jaas 配置。通过将最后一行中的 kafka-server-start.sh 编辑为：

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/etc/kafka/kafka_jaas_server.conf kafka.Kafka "$@"

Answer 1

我认为kafka0-8不支持sasl，从文档中很清楚只有kafka0-10具有sasl能力，而kafka0-10不支持Python。