構造化 ストリーミング + Kafka 統合ガイド (Kafkaブローカーバージョン 0.10.0 以上)

Structured Streaming integration for Kafka 0.10 to poll data from Kafka.

リンク

SBT/Maven プロジェクト定義を使用するScala/Javaアプリケーションのために、ストリーミングアプリケーションを以下のartifactとリンクします:

groupId = org.apache.spark
artifactId = spark-sql-kafka-0-10_2.11
version = 2.1.0

Pythonアプリケーションに関しては、アプリケーションをデプロイする時に、この上のライブラリとその依存物を追加する必要があるでしょう。以下の配備 サブセクションを見てください。

Kafkaソースストリームを作成

// Subscribe to 1 topic
val ds1 = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()
ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

// Subscribe to multiple topics
val ds2 = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1,topic2")
  .load()
ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

// Subscribe to a pattern
val ds3 = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribePattern", "topic.*")
  .load()
ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]
// Subscribe to 1 topic
Dataset<Row> ds1 = spark
  .readStream()
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()
ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

// Subscribe to multiple topics
Dataset<Row> ds2 = spark
  .readStream()
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1,topic2")
  .load()
ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

// Subscribe to a pattern
Dataset<Row> ds3 = spark
  .readStream()
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribePattern", "topic.*")
  .load()
ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
# Subscribe to 1 topic
ds1 = spark
  .readStream()
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()
ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

# Subscribe to multiple topics
ds2 = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1,topic2")
  .load()
ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

# Subscribe to a pattern
ds3 = spark
  .readStream()
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribePattern", "topic.*")
  .load()
ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

ソース内の各行は以下のスキーマを持ちます : <table class="table">

カラム種類 キー binary 値 binary topic string partition int offset long timestamp long timestampType int

</table>

以下のオプションはKafkaソースのために設定されなければなりません。

オプション値meaning
assign json string {"topicA":[0,1],"topicB":[2,4]} Specific TopicPartitions to consume. Kafkaソースについては、"assign", "subscribe" あるいは "subscribePattern" オプションのうちの1つだけが指定することができます。
購読 トピックのカンマ区切りのリスト 購読するトピックのリストKafkaソースについては、"assign", "subscribe" あるいは "subscribePattern" オプションのうちの1つだけが指定することができます。
subscribePattern Java regex 文字列 トピックを購読するために使われるパターン。Kafkaソースについては、"assign", "subscribe" あるいは "subscribePattern" オプションのうちの1つだけが指定することができます。
kafka.bootstrap.servers host:portのカンマ区切りのリスト Kafka "bootstrap.servers" 設定

以下の設定は任意です:

オプション値デフォルト:meaning
startingOffsets earliest, latest, or json string {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}} latest The start point when a query is started, either "earliest" which is from the earliest offsets, "latest" which is just from the latest offsets, or a json string specifying a starting offset for each TopicPartition. In the json, -2 as an offset can be used to refer to earliest, -1 to latest. Note: This only applies when a new Streaming query is started, and that resuming will always pick up from where the query left off. Newly discovered partitions during a query will start at earliest.
failOnDataLoss true or false true Whether to fail the query when it's possible that data is lost (e.g., topics are deleted, or offsets are out of range). これは間違ったアラームかも知れません。期待した通りに動作しない場合は無効にすることができます。
kafkaConsumer.pollTimeoutMs long 512 executor内のKafkaからデータをポーリングするタイムアウトのミリ秒。
fetchOffset.numRetries int 3 Kafkaの最新のオフセットを取り出すのを諦めるまでの試行回数。
fetchOffset.retryIntervalMs long 10 Kafkaのオフセットの取り出し試行までに待つミリ秒
maxOffsetsPerTrigger long none Rate limit on maximum number of offsets processed per trigger interval. The specified total number of offsets will be proportionally split across topicPartitions of different volume.

Kafka’s own configurations can be set via DataStreamReader.option with kafka. prefix, e.g, stream.option("kafka.bootstrap.servers", "host:port"). For possible kafkaParams, see Kafka consumer config docs.

Note that the following Kafka params cannot be set and the Kafka source will throw an exception:

配備

As with any Spark applications, spark-submit is used to launch your application. spark-sql-kafka-0-10_2.11 and its dependencies can be directly added to spark-submit using --packages, such as,

./bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0 ...

See Application Submission Guide for more details about submitting applications with external dependencies.

TOP
inserted by FC2 system