A side-by-side comparability of Apache Spark and Apache Flink for widespread streaming use circumstances

Apache Flink and Apache Spark are each open-source, distributed knowledge processing frameworks used extensively for large knowledge processing and analytics. Spark is thought for its ease of use, high-level APIs, and the power to course of giant quantities of information. Flink shines in its capability to deal with processing of information streams in real-time and low-latency stateful computations. Each help a wide range of programming languages, scalable options for dealing with giant quantities of information, and a variety of connectors. Traditionally, Spark began out as a batch-first framework and Flink started as a streaming-first framework.

On this put up, we share a comparative research of streaming patterns which might be generally used to construct stream processing functions, how they are often solved utilizing Spark (primarily Spark Structured Streaming) and Flink, and the minor variations of their strategy. Examples cowl code snippets in Python and SQL for each frameworks throughout three main themes: knowledge preparation, knowledge processing, and knowledge enrichment. In case you are a Spark person seeking to clear up your stream processing use circumstances utilizing Flink, this put up is for you. We don’t intend to cowl the selection of expertise between Spark and Flink as a result of it’s essential to judge each frameworks on your particular workload and the way the selection matches in your structure; somewhat, this put up highlights key variations to be used circumstances that each these applied sciences are generally thought-about for.

Apache Flink provides layered APIs that supply completely different ranges of expressiveness and management and are designed to focus on several types of use circumstances. The three layers of API are Course of Features (also called the Stateful Stream Processing API), DataStream, and Desk and SQL. The Stateful Stream Processing API requires writing verbose code however provides probably the most management over time and state, that are core ideas in stateful stream processing. The DataStream API helps Java, Scala, and Python and provides primitives for a lot of widespread stream processing operations, in addition to a stability between code verbosity or expressiveness and management. The Desk and SQL APIs are relational APIs that supply help for Java, Scala, Python, and SQL. They provide the best abstraction and intuitive, SQL-like declarative management over knowledge streams. Flink additionally permits seamless transition and switching throughout these APIs. To be taught extra about Flink’s layered APIs, consult with layered APIs.

Apache Spark Structured Streaming provides the Dataset and DataFrames APIs, which offer high-level declarative streaming APIs to symbolize static, bounded knowledge in addition to streaming, unbounded knowledge. Operations are supported in Scala, Java, Python, and R. Spark has a wealthy perform set and syntax with easy constructs for choice, aggregation, windowing, joins, and extra. You may also use the Streaming Desk API to learn tables as streaming DataFrames as an extension to the DataFrames API. Though it’s arduous to attract direct parallels between Flink and Spark throughout all stream processing constructs, at a really excessive stage, let’s imagine Spark Structured Streaming APIs are equal to Flink’s Desk and SQL APIs. Spark Structured Streaming, nonetheless, doesn’t but (on the time of this writing) provide an equal to the lower-level APIs in Flink that supply granular management of time and state.

Each Flink and Spark Structured Streaming (referenced as Spark henceforth) are evolving initiatives. The next desk supplies a easy comparability of Flink and Spark capabilities for widespread streaming primitives (as of this writing).

Within the following sections, we cowl the best widespread components in order that we will showcase how Spark customers can relate to Flink and vice versa. To be taught extra about Flink’s low-level APIs, consult with Course of Perform. For the sake of simplicity, we cowl the 4 use circumstances on this put up utilizing the Flink Desk API. We use a mixture of Python and SQL for an apples-to-apples comparability with Spark.

Knowledge preparation

On this part, we evaluate knowledge preparation strategies for Spark and Flink.

Studying knowledge

We first take a look at the only methods to learn knowledge from an information stream. The next sections assume the next schema for messages:

image: string,
value: int,
timestamp: timestamp,
company_info:
{
    title: string,
    employees_count: int
}

Studying knowledge from a supply in Spark Structured Streaming

In Spark Structured Streaming, we use a streaming DataFrame in Python that straight reads the info in JSON format:

spark = ...  # spark session

# specify schema
stock_ticker_schema = ...

# Create a streaming DataFrame
df = spark.readStream 
    .format("kafka") 
    .choice("kafka.bootstrap.servers", "mybroker1:port") 
    .choice("subject", "stock_ticker") 
    .load()
    .choose(from_json(col("worth"), stock_ticker_schema).alias("ticker_data")) 
    .choose(col("ticker_data.*"))

Be aware that now we have to produce a schema object that captures our inventory ticker schema (stock_ticker_schema). Examine this to the strategy for Flink within the subsequent part.

Studying knowledge from a supply utilizing Flink Desk API

For Flink, we use the SQL DDL assertion CREATE TABLE. You may specify the schema of the stream identical to you’ll any SQL desk. The WITH clause permits us to specify the connector to the info stream (Kafka on this case), the related properties for the connector, and knowledge format specs. See the next code:

# Create desk utilizing DDL

CREATE TABLE stock_ticker (
  image string,
  value INT,
  timestamp TIMESTAMP(3),
  company_info STRING,
  WATERMARK FOR timestamp AS timestamp - INTERVAL '3' MINUTE
) WITH (
 'connector' = 'kafka',
 'subject' = 'stock_ticker',
 'properties.bootstrap.servers' = 'mybroker1:port',
 'properties.group.id' = 'testGroup',
 'format' = 'json',
 'json.fail-on-missing-field' = 'false',
 'json.ignore-parse-errors' = 'true'
)

JSON flattening

JSON flattening is the method of changing a nested or hierarchical JSON object right into a flat, single-level construction. This converts a number of ranges of nesting into an object the place all of the keys and values are on the similar stage. Keys are mixed utilizing a delimiter reminiscent of a interval (.) or underscore (_) to indicate the unique hierarchy. JSON flattening is helpful when it’s essential to work with a extra simplified format. In each Spark and Flink, nested JSONs will be difficult to work with and may have extra processing or user-defined capabilities to control. Flattened JSONs can simplify processing and enhance efficiency as a result of decreased computational overhead, particularly with operations like complicated joins, aggregations, and windowing. As well as, flattened JSONs will help in simpler debugging and troubleshooting knowledge processing pipelines as a result of there are fewer ranges of nesting to navigate.

JSON flattening in Spark Structured Streaming

JSON flattening in Spark Structured Streaming requires you to make use of the choose methodology and specify the schema that you just want flattened. JSON flattening in Spark Structured Streaming entails specifying the nested subject title that you just’d like surfaced to the top-level listing of fields. Within the following instance, company_info is a nested subject and inside company_info, there’s a subject known as company_name. With the next question, we’re flattening company_info.title to company_name:

stock_ticker_df = ...  # Streaming DataFrame w/ schema proven above

stock_ticker_df.choose("image", "timestamp", "value", "company_info.title" as "company_name")

JSON flattening in Flink

In Flink SQL, you should use the JSON_VALUE perform. Be aware that you should use this perform solely in Flink variations equal to or larger than 1.14. See the next code:

SELECT
   image,
   timestamp,
   value,
   JSON_VALUE(company_info, 'lax $.title' DEFAULT NULL ON EMPTY) AS company_name
FROM
   stock_ticker

The time period lax within the previous question has to do with JSON path expression dealing with in Flink SQL. For extra data, consult with System (Constructed-in) Features.

Knowledge processing

Now that you’ve got learn the info, we will take a look at a number of widespread knowledge processing patterns.

Deduplication

Knowledge deduplication in stream processing is essential for sustaining knowledge high quality and guaranteeing consistency. It enhances effectivity by lowering the pressure on the processing from duplicate knowledge and helps with value financial savings on storage and processing.

Spark Streaming deduplication question

The next code snippet is said to a Spark Streaming DataFrame named stock_ticker. The code performs an operation to drop duplicate rows based mostly on the image column. The dropDuplicates methodology is used to get rid of duplicate rows in a DataFrame based mostly on a number of columns.

stock_ticker = ...  # Streaming DataFrame w/ schema proven above

stock_ticker.dropDuplicates("image")

Flink deduplication question

The next code exhibits the Flink SQL equal to deduplicate knowledge based mostly on the image column. The question retrieves the primary row for every distinct worth within the image column from the stock_ticker stream, based mostly on the ascending order of proctime:

SELECT image, timestamp, value
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY image ORDER BY proctime ASC) AS row_num
  FROM stock_ticker)
WHERE row_num = 1

Windowing

Windowing in streaming knowledge is a elementary assemble to course of knowledge inside specs. Home windows generally have time bounds, variety of information, or different standards. These time bounds bucketize steady unbounded knowledge streams into manageable chunks known as home windows for processing. Home windows assist in analyzing knowledge and gaining insights in actual time whereas sustaining processing effectivity. Analyses or operations are carried out on continually updating streaming knowledge inside a window.

There are two widespread time-based home windows used each in Spark Streaming and Flink that we’ll element on this put up: tumbling and sliding home windows. A tumbling window is a time-based window that may be a fastened measurement and doesn’t have any overlapping intervals. A sliding window is a time-based window that may be a fastened measurement and strikes ahead in fastened intervals that may be overlapping.

Spark Streaming tumbling window question

The next is a Spark Streaming tumbling window question with a window measurement of 10 minutes:

stock_ticker = ...  # Streaming DataFrame w/ schema proven above

# Get max inventory value in tumbling window
# of measurement 10 minutes
visitsByWindowAndUser = visits
   .withWatermark("timestamp", "3 minutes")
   .groupBy(
      window(stock_ticker.timestamp, "10 minutes"),
      stock_ticker.image)
   .max(stock_ticker.value)

Flink Streaming tumbling window question

The next is an equal tumbling window question in Flink with a window measurement of 10 minutes:

SELECT image, MAX(value)
  FROM TABLE(
    TUMBLE(TABLE stock_ticker, DESCRIPTOR(timestamp), INTERVAL '10' MINUTES))
  GROUP BY ticker;

Spark Streaming sliding window question

The next is a Spark Streaming sliding window question with a window measurement of 10 minutes and slide interval of 5 minutes:

stock_ticker = ...  # Streaming DataFrame w/ schema proven above

# Get max inventory value in sliding window
# of measurement 10 minutes and slide interval of measurement
# 5 minutes

visitsByWindowAndUser = visits
   .withWatermark("timestamp", "3 minutes")
   .groupBy(
      window(stock_ticker.timestamp, "10 minutes", "5 minutes"),
      stock_ticker.image)
   .max(stock_ticker.value)

Flink Streaming sliding window question

The next is a Flink sliding window question with a window measurement of 10 minutes and slide interval of 5 minutes:

SELECT image, MAX(value)
  FROM TABLE(
    HOP(TABLE stock_ticker, DESCRIPTOR(timestamp), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
  GROUP BY ticker;

Dealing with late knowledge

Each Spark Structured Streaming and Flink help occasion time processing, the place a subject throughout the payload can be utilized for outlining time home windows as distinct from the wall clock time of the machines doing the processing. Each Flink and Spark use watermarking for this function.

Watermarking is utilized in stream processing engines to deal with delays. A watermark is sort of a timer that units how lengthy the system can look ahead to late occasions. If an occasion arrives and is throughout the set time (watermark), the system will use it to replace a request. If it’s later than the watermark, the system will ignore it.

Within the previous windowing queries, you specify the lateness threshold in Spark utilizing the next code:

.withWatermark("timestamp", "3 minutes")

Which means that any information which might be 3 minutes late as tracked by the occasion time clock will probably be discarded.

In distinction, with the Flink Desk API, you’ll be able to specify a similar lateness threshold straight within the DDL:

WATERMARK FOR timestamp AS timestamp - INTERVAL '3' MINUTE

Be aware that Flink supplies extra constructs for specifying lateness throughout its numerous APIs.

Knowledge enrichment

On this part, we evaluate knowledge enrichment strategies with Spark and Flink.

Calling an exterior API

Calling exterior APIs from user-defined capabilities (UDFs) is analogous in Spark and Flink. Be aware that your UDF will probably be known as for each report processed, which can lead to the API getting known as at a really excessive request price. As well as, in manufacturing eventualities, your UDF code typically will get run in parallel throughout a number of nodes, additional amplifying the request price.

For the next code snippets, let’s assume that the exterior API name entails calling the perform:

response = my_external_api(request)

Exterior API name in Spark UDF

The next code makes use of Spark:

class Predict(ScalarFunction):
def open(self, function_context):

with open("assets.zip/assets/mannequin.pkl", "rb") as f:
self.mannequin = pickle.load(f)

def eval(self, x):
return self.mannequin.predict(x)

Exterior API name in Flink UDF

For Flink, assume we outline the UDF callExternalAPIUDF, which takes as enter the ticker image image and returns enriched details about the image by way of a REST endpoint. We will then register and name the UDF as follows:

callExternalAPIUDF = udf(callExternalAPIUDF(), result_type=DataTypes.STRING())

SELECT
    image, 
    callExternalAPIUDF(image) as enriched_symbol
FROM stock_ticker;

Flink UDFs present an initialization methodology that will get run one time (versus one time per report processed).

Be aware that it is best to use UDFs judiciously as an improperly applied UDF could cause your job to decelerate, trigger backpressure, and ultimately stall your stream processing software. It’s advisable to make use of UDFs asynchronously to keep up excessive throughput, particularly for I/O-bound use circumstances or when coping with exterior assets like databases or REST APIs. To be taught extra about how you should use asynchronous I/O with Apache Flink, consult with Enrich your knowledge stream asynchronously utilizing Amazon Kinesis Knowledge Analytics for Apache Flink.

Conclusion

Apache Flink and Apache Spark are each quickly evolving initiatives and supply a quick and environment friendly option to course of huge knowledge. This put up targeted on the highest use circumstances we generally encountered when prospects wished to see parallels between the 2 applied sciences for constructing real-time stream processing functions. We’ve included samples that had been most often requested on the time of this writing. Tell us if you happen to’d like extra examples within the feedback part.

Concerning the writer

Deepthi Mohan is a Principal Product Supervisor on the Amazon Kinesis Knowledge Analytics staff.

Karthi Thyagarajan was a Principal Options Architect on the Amazon Kinesis staff.