Reliable Sources Must Be Replayable

In Structured Streaming, replayability is the capacity to request a part of the stream that had been already requested but not committed yet. Just like we can rewind that Netflix series we are watching to see a piece we just missed because of a distraction, sources must offer the capability to replay a piece of the stream that was already requested but not committed. This is done by calling getBatch with the offset range that we want to receive again.

A source is considered to be reliable when it can produce an uncommitted offset range even after a total failure of the Structured Streaming process. In this failure recovery process, offsets are restored from their last known checkpoint and are requested again from the source. This requires the actual streaming system that is backing the source implementation to store data safely outside the streaming process. By requiring replayability from the source, Structured Streaming delegates recovery responsibility to the source. This implies that only reliable sources work with Structured Streaming to create strong end-to-end delivery guarantees.

Sources Must Provide a Schema

A defining characteristic of the structured APIs of Spark is that they rely on schema information to handle the data at different levels. As opposed to processing opaque Strings or Byte Array blobs, schema information provides insights on how the data is shaped in terms of fields and types. We can use schema information to drive optimizations at different levels in the stack, from query planning to the internal binary representation of data, storage, and access to it.

Sources must provide schema information that describes the data they produce. Some source implementations allow this schema to be configured and use this configuration information to automatically parse incoming data and transform it into valid records. In fact, many file-based streaming sources such as JSON or comma-separated values (CSV) files follow this model, in which the user must provide the schema used by the file format to ensure proper parsing. Some other sources use a fixed internal schema that expresses the metadata information of every record and leaves the parsing of the payload to the application.

From an architectural perspective, creating schema-driven streaming applications is desirable because it facilitates the global understanding of how data flows through the system and drives the formalization of the different stages of a multiprocess streaming pipeline.

import org.apache.spark.sql.Encoders

// Define the case class hierarchy
case class Coordinates(latitude: Double, longitude: Double)
case class Vehicle(id: String, `type`: String, location: Coordinates )
// Obtain the Encoder, and the schema from the Encoder
val schema = Encoders.product[Vehicle].schema

val sample = spark.read.parquet(<path-to-sample>)
val schema = sample.schema

Specifying a File Format

The files are read using a specified format, which is provided with the .format(<format_name>) method in the readStream builder, or by using the dedicated methods in the DataStreamReader that indicate the format to use; for example, readStream.parquet('/path/to/dir/'). When using the dedicated methods corresponding to each supported format, the method call should be done as the last call of the builder.

For example, the three forms in Example 10-1 are equivalent.

// Use format and load path
val fileStream = spark.readStream
  .format("parquet")
  .schema(schema)
  .load("hdfs://data/exchange")

// Use format and path options
val fileStream = spark.readStream
  .format("parquet")
  .option("path", "hdfs://data/exchange")
  .schema(schema)
  .load()

// Use dedicated method
val fileStream = spark.readStream
  .schema(schema)
  .parquet("hdfs://data/exchange")

As of Spark v2.3.0, the following file-based formats are supported by Structured Streaming. These are the same file formats supported by the static DataFrame, Dataset, and SQL APIs:

• CSV

• JSON

• Parquet

• ORC

• text

• textFile

Common Options

Regardless of the specific format, the general functionality of the File source is to monitor a directory in a shared filesystem identified by its specific URL. All file formats support a common set of options that control the file inflow and define the aging criteria of the files.

These options can be set for all file-based sources:

maxFilesPerTrigger (Default: unset)

Indicates how many files will be consumed at each query trigger. This setting limits the number of files processed at each trigger, and in doing so, it helps to control the data inflow in the system.

latestFirst (Default: false)

When this flag is set to true, newer files are elected for processing first. Use this option when the most recent data has higher priority over older data.

maxFileAge (Default: 7 days)

Defines an age threshold for the files in the directory. Files older than the threshold will not be eligible for processing and will be effectively ignored. This threshold is relative to the most recent file in the directory and not to the system clock. For example, if maxFileAge is 2 days and the most recent file is from yesterday, the threshold to consider a file too old will be older than three days ago. This dynamic is similar to watermarks on event time.

fileNameOnly (Default: false)

When set to true, two files will be considered the same if they have the same name; otherwise, the full path will be considered.

Common Text Parsing Options (CSV, JSON)

Some file formats, such as CSV and JSON, use a configurable parser to transform the text data in each file into structured records. It’s possible for upstream processes to create records that do not fulfill the expected format. These records are considered corrupted.

Streaming systems are characterized by their continuous running. A streaming process should not fail when bad data is received. Depending on the business requirements, we can either drop the invalid records or route the data considered corrupted to a separate error-handling flow.

JSON File Source Format

The JSON format support for the File source lets us consume text files encoded as JSON, in which each line in the file is expected to be a valid JSON object. The JSON records are parsed using the provided schema. Records that do not follow the schema are considered invalid, and there are several options available to control the handling of invalid records.

// valid record
{"firstname":"Alice", 'lastname': 'Wonderland'}
// invalid nesting
{"firstname":'Elvis "The King"', 'lastname': 'Presley'}
// correct escaping
{"firstname":'Elvis "The King"', 'lastname': 'Presley'}

CSV File Source Format

CSV is a popular tabular data storage and exchange format that is widely supported by enterprise applications. The File Source CSV format support allows us to ingest and process the output of such applications in a Structured Streaming application. Although the name “CSV” originally indicated that values are separated by commas, often the separation character can be freely configured. There are many configuration options available to control the way data is transformed from plain text to structured records.

In the rest of this section, we cover the most common options and, in particular, those that are relevant for the streaming process. For formatting-related options, refer to the latest documentation. These are the most commonly used CSV parsing options:

#Timestamps are in ISO 8601 compliant format
x097abba, 2018-04-01T16:32:56+00:00, 55
x053ba0bab, 2018-04-01T16:35:02+00:00, 32

Parquet File Source Format

Apache Parquet is a column-oriented, file-based data storage format. The internal representation splits original rows into chunks of columns that are stored using compression techniques. As a consequence, queries that require specific columns do not need to read the complete file. Instead, the relevant pieces can be independently addressed and retrieved. Parquet supports complex nested data structures and preserves the schema structure of the data. Due to its enhanced query capabilities, efficient use of storage space and the preservation of schema information, Parquet is a popular format for storing large, complex datasets.

// Use format and load path
val fileStream = spark.readStream
  .schema(schema)
  .parquet("hdfs://data/folder")

Text File Source Format

The text format for the File source supports the ingestion of plain-text files. Using configuration options, it’s possible to ingest the text either line by line or the entire file as a single text blob. The schema of the data produced by this source is naturally StringType and does not need to be specified. This is a generic format that we can use to ingest arbitrary text for further programmatic processing, from the famous word count to custom parsing of proprietary text formats.

// Text specified as format
>val fileStream = spark.readStream.format("text").load("hdfs://data/folder")
fileStream: org.apache.spark.sql.DataFrame = [value: string]

// Text specified through dedicated method
>val fileStream = spark.readStream.text("hdfs://data/folder")
fileStream: org.apache.spark.sql.DataFrame = [value: string]

// TextFile specified through dedicated method
val fileStream = spark.readStream.textFile("/tmp/data/stream")
fileStream: org.apache.spark.sql.Dataset[String] = [value: string]

Setting Up a Kafka Source

To create a Kafka source, we use the format("kafka") method with the createStream builder on the Spark Session. We need two mandatory parameters to connect to Kafka: the addresses of the Kafka brokers, and the topic(s) to which we want to connect.

The address of the Kafka brokers to connect to is provided through the option kafka.bootstrap.servers as a String containing a comma-separated list of host:port pairs.

Example 10-6 presents a simple Kafka source definition. It subscribes to a single topic, topic1, by connecting to the brokers located at host1:port1, host2:port2, and host3:port3.

>val kafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2,host3:port3")
  .option("subscribe", "topic1")
  .option("checkpointLocation", "hdfs://spark/streaming/checkpoint")
  .load()

kafkaStream: org.apache.spark.sql.DataFrame =
  [key: binary, value: binary ... 5 more fields]

>kafkaStream.printSchema

root
 |-- key: binary (nullable = true)
 |-- value: binary (nullable = true)
 |-- topic: string (nullable = true)
 |-- partition: integer (nullable = true)
 |-- offset: long (nullable = true)
 |-- timestamp: timestamp (nullable = true)
 |-- timestampType: integer (nullable = true)

>val dataStream = kafkaStream.selectExpr("CAST(key AS STRING)",
                                         "CAST(value AS STRING)")
                             .as[(String, String)]

dataStream: org.apache.spark.sql.Dataset[(String, String)] =
  [key: string, value: string]

The result of that call is a DataFrame with five fields: key, value, topic, partition, offset, timestamp, timestampType. Having a schema consisting of these five fields is fixed the Kafka source. It provides the raw key and values from Kafka and the metadata of each consumed record.

Usually, we are interested in only the key and value of the message. Both the key and the value contain a binary payload, represented internally as a Byte Array. When the data is written to Kafka using a String serializer, we can read that data back by casting the values to String as we have done in the last expression in the example. Although text-based encoding is a common practice, it’s not the most space-efficient way of exchanging data. Other encodings, such as the schema-aware AVRO format, might offer a better space efficiency with the added benefit of embedding the schema information.

The additional metadata in the message, such as topic, partition, or offset can be used in more complex scenarios. For example, the topic field will contain the topic that produced the record and could be used as a label or discriminator, in case we subscribe to several topics at the same time.

Selecting a Topic Subscription Method

There are three different ways to specify the topic or topics that we want to consume:

• subscribe

• subscribePattern

• assign

The Kafka source setup must contain one and only one of these subscription options. They provide different levels of flexibility to select the topic(s) and even the partitions to subscribe to:

subscribe

Takes a single topic or a list of comma-separated topics: topic1, topic2, ..., topicn. This method subscribes to each topic and creates a single, unified stream with the data of the union of all the topics; for example, .option("subscribe", "topic1,topic3").

subscribePattern

This is similar to subscribe in behavior, but the topics are specified with a regular expression pattern. For example, if we have topics 'factory1Sensors', 'factory2Sensors', 'street1Sensors', 'street2Sensors', we can subscribe to all “factory” sensors with the expression .option("subscribePattern", "factory[\d]+Sensors").

assign

Allows the fine-grained specification of specific partitions per topic to consume. This is known as TopicPartitions in the Kafka API. The partitions per topic are indicated using a JSON object, in which each key is a topic and its value is an array of partitions. For example, the option definition .option("assign", """{"sensors":[0,1,3]}""") would subscribe to the partitions 0, 1, and 3 of topic sensors. To use this method we need information about the topic partitioning. We can obtain the partition information programmatically by using the Kafka API or through configuration.

Configuring Kafka Source Options

There are two categories of configuration options for the Structured Streaming source for Kafka: dedicated source configuration, and pass-through options that are given directly to the underlying Kafka consumer.

Kafka Consumer Options

It’s possible to pass configuration options through to the underlying Kafka consumer of this source. We do this by adding a 'kafka.' prefix to the configuration key that we want to set.

For example, to configure Transport Layer Security (TLS) options for the Kafka source, we can set the Kafka consumer configuration option security.protocol by setting kafka.security.protocol in the source configuration.

Example 10-7 demonstrates how to configure TLS for the Kafka source using this method.

val tlsKafkaSource = spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1, host2:port2")
  .option("subscribe", "topsecret")
  .option("kafka.security.protocol", "SSL")
  .option("kafka.ssl.truststore.location", "/path/to/truststore.jks")
  .option("kafka.ssl.truststore.password", "truststore-password")
  .option("kafka.ssl.keystore.location", "/path/to/keystore.jks")
  .option("kafka.ssl.keystore.password", "keystore-password")
  .option("kafka.ssl.key.password", "password")
  .load()

Configuration

To connect to a TCP server, we need the address of the host and a port number. It’s also possible to configure the Socket source to add the timestamp at which each line of data is received.

These are the configuration options:

host (mandatory)

The DNS hostname or IP address of the TCP server to which to connect.

port (mandatory)

The port number of the TCP server to which to connect.

includeTimestamp (default: false)

When enabled, the Socket source adds the timestamp of arrival to each line of data. It also changes the schema produced by this source, adding the timestamp as an additional field.

In Example 10-8, we observe the two modes of operation that this source offers. With the host, port configuration, the resulting streaming DataFrame has a single field named value of type String. When we add the flag includeTimestamp set to true, the schema of the resulting streaming DataFrame contains the fields value and timestamp, where value is of type String as before, and timestamp is of type Timestamp. Also, observe the log warning this source prints at creation.

// Only using host and port

>val stream = spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9876)
  .load()

18/04/14 17:02:34 WARN TextSocketSourceProvider:
The socket source should not be used for production applications!
It does not support recovery.

stream: org.apache.spark.sql.DataFrame = [value: string]

// With added timestamp information

val stream = spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9876)
  .option("includeTimestamp", true)
  .load()

18/04/14 17:06:47 WARN TextSocketSourceProvider:
The socket source should not be used for production applications!
It does not support recovery.

stream: org.apache.spark.sql.DataFrame = [value: string, timestamp: timestamp]

Operations

The Socket source creates a TCP client that connects to a TCP server specified in the configuration. This client runs on the Spark driver. It keeps the incoming data in memory until it is consumed by the query and the corresponding offset is committed. The data from committed offsets is evicted, keeping the memory usage stable under normal circumstances.

Recall from the discussion in “Understanding Sources”, a source is considered reliable if it can replay an uncommitted offset even in the event of failure and restart of the streaming process. This source is not considered to be reliable, because a failure of the Spark driver will result in losing all uncommitted data in memory.

This source should be used only for cases in which data loss is acceptable for the use case.

Options

The Rate source supports a few options that control the throughput and level of parallelism:

rowsPerSecond (default: 1)

The number of rows to generate each second.

rampUpTime (default: 0)

At the start of the stream, the generation of records will increase progressively until this time has been reached. The increase is linear.

numPartitions (default: default spark parallelism level)

The number of partitions to generate. More partitions increase the parallelism level of the record generation and downstream query processing.