Introduction

The streaming benchmark consists of the following parts:

Data Generation

The role of data generator is to generate a steady stream of data and send it to the Kafka cluster. Each record is labeled with the timestamp when it is generated.

Kafka cluster

Kafka is a messaging system. Each streaming workload saves the data to its own topics.

Test cluster

This could be a Spark cluster, Flink cluster, Storm cluster or Gearpump cluster. The streaming application (identity, repartition, wordcount, fixwindow) reads data from Kafka, processes the data and writes the results back to another topic in Kafka. Each record in the result is also labeled a timestamp.

Metrics reader

It reads the result in Kafka and calculate the time difference(Record out time - Record in time) and generate the report.

Setup

Python 2.x(>=2.6) is required.

Supported Hadoop version: Apache Hadoop 2.x, CDH5.x, HDP
Supported Spark version: 1.6.x, 2.0.x, 2.1.x, 2.2.x
Build HiBench according to build HiBench.
Start HDFS, Yarn in the cluster.
Setup ZooKeeper (3.4.8 is preferred): Start doc
Setup Apache Kafka (0.8.2.2, scala version 2.10 is preferred): Start Kafka
Setup one of the streaming frameworks that you want to test.
- Apache Spark (1.6.1, is preferred)
- Apache Storm (1.0.1 is preferred).
- Apache Flink (1.0.3 is prefered).
- Apache Gearpump (0.8.1 is prefered)

We use Apache Spark 2.2.1 to streaming frameworks.

Hadoop Configuration

Hadoop is used to generate the input data of the workloads. Create and edit conf/hadoop.conf：

cp conf/hadoop.conf.template conf/hadoop.conf

Set the below properties properly:

Property	Meaning
hibench.hadoop.home	/root/hadoop
hibench.hadoop.executable	${hibench.hadoop.home}/bin/hadoop
hibench.hadoop.configure.dir	${hibench.hadoop.home}/etc/hadoop
hibench.hdfs.master	hdfs://wls-arm-huawei01:9000
hibench.hadoop.release	apache, cdh5, hdp

Note: For CDH and HDP users, please update hibench.hadoop.executable, hibench.hadoop.configure.dir and hibench.hadoop.release properly.

The default value is for Apache release.

Kafka Configuration

Set the below Kafka properites in conf/hibench.conf and leave others as default.

Property	Meaning
hibench.streambench.kafka.home	/root/kafka_2.10-0.8.2.2
hibench.streambench.zkHost	wls-arm-huawei01:2181
hibench.streambench.kafka.brokerList	wls-arm-huawei01:9092
hibench.streambench.kafka.topicPartitions	Number of partitions of generated topic (default 20)

Generate the data

Take workload identity as an example. genSeedDataset.sh generates the seed data on HDFS. dataGen.sh sends the data to Kafka.

bin/workloads/streaming/identity/prepare/genSeedDataset.sh
bin/workloads/streaming/identity/prepare/dataGen.sh

Run the streaming application

While the data are being sent to the Kafka, start the streaming application. Take Spark streaming as an example.

bin/workloads/streaming/identity/spark/run.sh

Generate the report

metrics_reader.sh is used to generate the report.

bin/workloads/streaming/identity/common/metrics_reader.sh

Debug for running Streaming bench

when running the streaming application:

bin/workloads/streaming/identity/spark/run.sh

There are some issues as follows:

streaming identity run into ERROR CheckpointWriter: Could not submit checkpoint task to the thread pool executor

root@wls-arm-huawei01:~/HiBench# bin/workloads/streaming/identity/spark/run.sh
patching args=
Parsing conf: /root/HiBench/conf/hadoop.conf
Parsing conf: /root/HiBench/conf/hibench.conf
Parsing conf: /root/HiBench/conf/spark.conf
Parsing conf: /root/HiBench/conf/workloads/streaming/identity.conf
probe sleep jar: /root/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-2.7.1-tests.jar
start SparkStreamingIdentity bench
Export env: SPARKBENCH_PROPERTIES_FILES=/root/HiBench/report/identity/spark/conf/sparkbench/sparkbench.conf
Export env: HADOOP_CONF_DIR=/root/hadoop/etc/hadoop
Submit Spark job: /root/spark/bin/spark-submit  --properties-file /root/HiBench/report/identity/spark/conf/sparkbench/spark.conf --class com.intel.hibench.sparkbench.streaming.RunBench --master yarn-client --num-executors 2 --executor-cores 4 --executor-memory 6g /root/HiBench/sparkbench/assembly/target/sparkbench-assembly-7.1-SNAPSHOT-dist.jar /root/HiBench/report/identity/spark/conf/sparkbench/sparkbench.conf
Warning: Master yarn-client is deprecated since 2.0. Please use master "yarn" with specified deploy mode instead.
metrics is being written to kafka topic SPARK_identity_1_5_50_1528365200903
Reporter Topic: SPARK_identity_1_5_50_1528365200903
18/06/07 17:53:20 INFO zkclient.ZkEventThread: Starting ZkClient event thread.
18/06/07 17:53:20 INFO zookeeper.ZooKeeper: Client environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT
18/06/07 17:53:20 INFO zookeeper.ZooKeeper: Client environment:host.name=wls-x86-hp05
18/06/07 17:53:20 INFO zookeeper.ZooKeeper: Client environment:java.version=1.8.0_171
........
.....
...
..


18/06/07 17:53:34 INFO cluster.YarnClientSchedulerBackend: Application application_1528275127402_0017 has started running.
18/06/07 17:53:34 INFO util.Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 45199.
18/06/07 17:53:34 INFO netty.NettyBlockTransferService: Server created on 10.169.41.13:45199
18/06/07 17:53:34 INFO storage.BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
18/06/07 17:53:34 INFO storage.BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 10.169.41.13, 45199, None)
18/06/07 17:53:34 INFO storage.BlockManagerMasterEndpoint: Registering block manager 10.169.41.13:45199 with 3.0 GB RAM, BlockManagerId(driver, 10.169.41.13, 45199, None)
18/06/07 17:53:34 INFO storage.BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 10.169.41.13, 45199, None)
18/06/07 17:53:34 INFO storage.BlockManager: Initialized BlockManager: BlockManagerId(driver, 10.169.41.13, 45199, None)
18/06/07 17:53:35 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@3fb450d7{/metrics/json,null,AVAILABLE,@Spark}
18/06/07 17:53:37 INFO cluster.YarnSchedulerBackend$YarnDriverEndpoint: Registered executor NettyRpcEndpointRef(spark-client://Executor) (10.169.41.13:43928) with ID 1
18/06/07 17:53:37 INFO storage.BlockManagerMasterEndpoint: Registering block manager wls-x86-hp05:44621 with 3.0 GB RAM, BlockManagerId(1, wls-x86-hp05, 44621, None)
18/06/07 17:53:52 INFO cluster.YarnClientSchedulerBackend: SchedulerBackend is ready for scheduling beginning after waiting maxRegisteredResourcesWaitingTime: 30000(ms)
java.util.concurrent.RejectedExecutionException: Task org.apache.spark.streaming.CheckpointWriter$CheckpointWriteHandler@76efdbf9 rejected from java.util.concurrent.ThreadPoolExecut...
        at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
        at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
        at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
        at org.apache.spark.streaming.CheckpointWriter.write(Checkpoint.scala:287)
        at org.apache.spark.streaming.scheduler.JobGenerator.doCheckpoint(JobGenerator.scala:297)
        at org.apache.spark.streaming.scheduler.JobGenerator.org$apache$spark$streaming$scheduler$JobGenerator$$processEvent(JobGenerator.scala:186)
        at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:89)
        at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:88)
        at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
java.util.concurrent.RejectedExecutionException: Task org.apache.spark.streaming.CheckpointWriter$CheckpointWriteHandler@54d5eb6a rejected from java.util.concurrent.ThreadPoolExecut...
        at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
        at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
        at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
        at org.apache.spark.streaming.CheckpointWriter.write(Checkpoint.scala:287)
        at org.apache.spark.streaming.scheduler.JobGenerator.doCheckpoint(JobGenerator.scala:297)
        at org.apache.spark.streaming.scheduler.JobGenerator.org$apache$spark$streaming$scheduler$JobGenerator$$processEvent(JobGenerator.scala:186)
        at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:89)
        at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:88)
        at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
.....
...
..
.

The solutions:

Turn off checkpointing :

hibench.streambench.spark.checkpointPath false

Set "conf/spark.conf:hibench.streambench.spark.batchInterval" to a larger value

2. Process of generating data hangs

The solutions:

Modify conf/hibench.conf to make data-generating not in infinity mode:

# Total round count of data send (default: -1 means infinity)
hibench.streambench.datagen.totalRounds         -1   ->  512
# Number of total records that will be generated (default: -1 means infinity)
hibench.streambench.datagen.totalRecords        -1   ->  1024

Spark streaming bench report

Identity:

Arch	count	throughput(msgs/s)	max_latency(ms)	mean_latency(ms)	min_latency(ms)	stddev_latency(ms)	p50_latency(ms)	p75_latency(ms)	p95_latency(ms)	p98_latency(ms)	p99_latency(ms)	p999_latency(ms)
Arm64	2015	99	2339	338.531	54	437.862	210	307	1579	1985.32	2148	2338
X86	6115	100	2269	162.566	25	349.292	80	87	974	1687.68	1817	2220

Repartition:

Arch	count	throughput(msgs/s)	max_latency(ms)	mean_latency(ms)	min_latency(ms)	stddev_latency(ms)	p50_latency(ms)	p75_latency(ms)	p95_latency(ms)	p98_latency(ms)	p99_latency(ms)	p999_latency(ms)
Arm64	2160	99	2896	458.714	89	573.476	257	351	1961.55	2447.12	2678.02	2885.433
X86	2930	99	2720	569.22	73	828.736vi	132	503	2480.45	2549.38	2573.07	2674.243

Wordcount:

Arch	count	throughput(msgs/s)	max_latency(ms)	mean_latency(ms)	min_latency(ms)	stddev_latency(ms)	p50_latency(ms)	p75_latency(ms)	p95_latency(ms)	p98_latency(ms)	p99_latency(ms)	p999_latency(ms)
Arm64	4145	99	3495	967.401	48	844.909	743	1482.5	2695.7	3005.36	3134.54	3445.854
X86	5674	92	4825	3879.146	2534	428.128	3901	4171.25	4586	4682	4731.5	4818

Big Data Team Space

Spark Streaming benchmark on Arm64

Introduction

Setup

Hadoop Configuration

Set the below properties properly:

Kafka Configuration

Generate the data

Run the streaming application

Generate the report

Debug for running Streaming bench

The solutions:

The solutions:

Spark streaming bench report

Identity:

Repartition:

Wordcount: