如何将 Spark 数据帧写入 Kinesis Stream？

Question

我正在使用 Spark Streaming 从 kafka 主题创建一个 Dataframe。我想将 Dataframe 写入 Kinesis Producer。据我所知，目前还没有官方的 API。但互联网上有多个可用的 API，但遗憾的是，它们都不适合我。火花版本：2.2 斯卡拉：2.11

我尝试使用https://github.com/awslabs/kinesis-kafka-connector并构建 jar。但由于此 jar 和 Spark API 之间的包名称冲突而出现错误。请帮忙。

########## 这是其他人的代码：

spark-shell --jars spark-sql-kinesis_2.11-2.2.0.jar,spark-sql-kafka-0-10_2.11-2.1.0.jar,spark-streaming-kafka-0-10-assembly_2.10-2.1.0.jar --files kafka_client_jaas_spark.conf --properties-file gobblin_migration.conf --conf spark.port.maxRetries=100 --driver-java-options "-Djava.security.auth.login.config=kafka_client_jaas_spark.conf" --conf "spark.driver.extraJavaOptions=-Djava.security.auth.login.config=kafka_client_jaas_spark.conf" --conf "spark.executor.extraJavaOptions=-Djava.security.auth.login.config=kafka_client_jaas_spark.conf"

import java.io.File
import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.spark.SparkException
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
import scala.sys.process._
import org.apache.log4j.{ Logger, Level, LogManager, PropertyConfigurator }
import org.apache.spark.sql.streaming.Trigger

val streamingInputDF =spark.readStream.format("kafka").option("kafka.bootstrap.servers","bootstrap server").option("subscribe", "<kafkatopic>").option("startingOffsets", "latest").option("failOnDataLoss", "false").option("kafka.security.protocol", "SASL_PLAINTEXT").load()

val xdf=streamingInputDF.select(col("partition").cast("String").alias("partitionKey"),col("value").alias("data"))

xdf.writeStream.format("kinesis").option("checkpointLocation", "<hdfspath>").outputMode("Append").option("streamName", "kinesisstreamname").option("endpointUrl","kinesisendpoint").option("awsAccessKeyId", "accesskey").option("awsSecretKey","secretkey").start().awaitTermination()

对于 jar spark-sql-kinesis_2.11-2.2.0.jar，请转到quoble，下载适合您的 Spark 版本的包，构建 jar。

如果您位于公司网络后面，请在启动 Spark 之前设置代理。导出 http_proxy= http://服务器 IP:端口/ 导出 https_proxy= https://服务器 IP:端口/

Answer 1

Kafka Connect 是一项服务，您可以向其中发布连接器规范（在本例中为 kinesis），然后由该服务负责运行连接器。在处理记录时它还支持相当多的转换。Kafka Connect 插件不适用于 Spark 应用程序。

如果您的用例要求您在处理记录时执行一些业务逻辑，那么您可以使用 Spark Streaming 或 Structured Streaming 方法。

如果你想采用基于 Spark 的方法，下面是我能想到的 2 个选项。

1. 使用结构化流。您可以使用 Kinesis 的结构化流连接器。您可以在这里找到一个。可能还有其他人。这是我所知道的唯一稳定的开源连接器。您可以在此处找到使用 Kinesis 作为接收器的示例。

2. 使用Kinesis Producer Library或aws-java-sdk-kinesis库从 Spark Streaming 应用程序发布记录。使用 KPL 是此处的首选方法。您可以mapPartitions为每个分区创建一个 Kinesis 客户端，并使用这些库发布记录。AWS 文档中有很多关于这两个库的示例。