Spark源码系列之spark2.2的StructuredStreaming使用及源码介绍

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
  <version>2.2.0</version>
</dependency>

val spark = SparkSession
  .builder()
  .appName("Spark structured streaming Kafka example")
  .master("local")
  .getOrCreate()
val inputstream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "blog")
  .load()
val keyValueString = inputstream.selectExpr("CAST(key AS STRING)", "CAST( value AS STRING)").as[(String, String)]
val wordCounts = keyValueString.map(_._2.split(" ")).groupBy("value").count()
val query = wordCounts.writeStream.trigger(Trigger.ProcessingTime(1000000))
  .outputMode("complete")
  .format("console")
  .start()
query.awaitTermination()

noAggDF
  .writeStream
  .format("parquet")
  .option("checkpointLocation", "path/to/checkpoint/dir")
  .option("path", "path/to/destination/dir")
  .start()

writeStream
  .foreach(...)
.start()

aggDF
  .writeStream
  .outputMode("complete")
  .format("console")
  .start()

// Have all the aggregates in an in-memory table
aggDF
  .writeStream
  .queryName("aggregates")    // this query name will be the table name
 .outputMode("complete")
  .format("memory")
  .start()
spark.sql("select * from aggregates").show()

val ds = df
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "topic1")
  .start()

val dataSource =
 DataSource(
    sparkSession,
 userSpecifiedSchema = userSpecifiedSchema,
 className = source,
 options = extraOptions.toMap)
Dataset.ofRows(sparkSession, StreamingRelation(dataSource))

object StreamingRelation {
 def apply(dataSource: DataSource): StreamingRelation = {
    StreamingRelation(
      dataSource, dataSource.sourceInfo.name, dataSource.sourceInfo.schema.toAttributes)
  }
}

//加载所有DataSourceRegister的实现
val serviceLoader = ServiceLoader.load(classOf[DataSourceRegister], loader)
try {
 serviceLoader.asScala.filter(_.shortName().equalsIgnoreCase(provider1)).toList

providingClass.newInstance() match {
 case s: StreamSourceProvider =>
 val (name, schema) = s.sourceSchema(
      sparkSession.sqlContext, userSpecifiedSchema, className, caseInsensitiveOptions)
 SourceInfo(name, schema, Nil)

val (useTempCheckpointLocation, recoverFromCheckpointLocation) =
 if (source == "console") {
    (true, false)
  } else {
    (false, true)
  }
val dataSource =
 DataSource(
 df.sparkSession,
 className = source,
 options = extraOptions.toMap,
 partitionColumns = normalizedParCols.getOrElse(Nil))
df.sparkSession.sessionState.streamingQueryManager.startQuery(
 extraOptions.get("queryName"),
 extraOptions.get("checkpointLocation"),
 df,
 dataSource.createSink(outputMode),
 outputMode,
 useTempCheckpointLocation = useTempCheckpointLocation,
 recoverFromCheckpointLocation = recoverFromCheckpointLocation,
 trigger = trigger)

providingClass.newInstance() match {
 case s: StreamSinkProvider =>
    s.createSink(sparkSession.sqlContext, caseInsensitiveOptions, partitionColumns, outputMode)

val query = createQuery(
  userSpecifiedName,
 userSpecifiedCheckpointLocation,
 df,
 sink,
 outputMode,
 useTempCheckpointLocation,
 recoverFromCheckpointLocation,
 trigger,
 triggerClock)

new StreamingQueryWrapper(new StreamExecution(
  sparkSession,
 userSpecifiedName.orNull,
 checkpointLocation,
 analyzedPlan,
 sink,
 trigger,
 triggerClock,
 outputMode,
 deleteCheckpointOnStop))

val microBatchThread =
 new StreamExecutionThread(s"stream execution thread for $prettyIdString") {
 override def run(): Unit = {
 // To fix call site like "run at <unknown>:0", we bridge the call site from the caller
      // thread to this micro batch thread
 sparkSession.sparkContext.setCallSite(callSite)
      runBatches()
    }
  }

private val triggerExecutor = trigger match {
 case t: ProcessingTime => ProcessingTimeExecutor(t, triggerClock)
 case OneTimeTrigger => OneTimeExecutor()
 case _ => throw new IllegalStateException(s"Unknown type of trigger: $trigger")
}

if (dataAvailable) {
 currentStatus = currentStatus.copy(isDataAvailable = true)
  updateStatusMessage("Processing new data")
  runBatch(sparkSessionToRunBatches)
}

// Request unprocessed data from all sources.
newData = reportTimeTaken("getBatch") {
 availableOffsets.flatMap {
 case (source, available)
 if committedOffsets.get(source).map(_ != available).getOrElse(true) =>
 val current = committedOffsets.get(source)
 val batch = source.getBatch(current, available)
      logDebug(s"Retrieving data from $source: $current -> $available")
 Some(source -> batch)
 case _ => None
  }
}

// Replace sources in the logical plan with data that has arrived since the last batch.
val withNewSources = logicalPlan transform {
 case StreamingExecutionRelation(source, output) =>
 newData.get(source).map { data =>
 val newPlan = data.logicalPlan
 assert(output.size == newPlan.output.size,
 s"Invalid batch: ${Utils.truncatedString(output, ",")} != " +
 s"${Utils.truncatedString(newPlan.output, ",")}")
      replacements ++= output.zip(newPlan.output)
      newPlan
    }.getOrElse {
 LocalRelation(output)
    }
}

// Rewire the plan to use the new attributes that were returned by the source.
val replacementMap = AttributeMap(replacements)
val triggerLogicalPlan = withNewSources transformAllExpressions {
 case a: Attribute if replacementMap.contains(a) => replacementMap(a)
 case ct: CurrentTimestamp =>
 CurrentBatchTimestamp(offsetSeqMetadata.batchTimestampMs,
 ct.dataType)
 case cd: CurrentDate =>
 CurrentBatchTimestamp(offsetSeqMetadata.batchTimestampMs,
 cd.dataType, cd.timeZoneId)
}

reportTimeTaken("queryPlanning") {
 lastExecution = new IncrementalExecution(
    sparkSessionToRunBatch,
 triggerLogicalPlan,
 outputMode,
 checkpointFile("state"),
 currentBatchId,
 offsetSeqMetadata)
 lastExecution.executedPlan // Force the lazy generation of execution plan
}

val nextBatch =
 new Dataset(sparkSessionToRunBatch, lastExecution, RowEncoder(lastExecution.analyzed.schema))

reportTimeTaken("addBatch") {
  sink.addBatch(currentBatchId, nextBatch)
}

data.sparkSession.createDataFrame(
 data.sparkSession.sparkContext.parallelize(data.collect()), data.schema)
  .show(numRowsToShow, isTruncated)