Scenario

Scenario

Assume that you need to collect statistics on the number of events in each session and the start and end timestamp of the sessions.

You need to export the sessions that are in the updated state in this batch.

Data Planning

1. Generate simulated data in Kafka (the Kafka permission is required).

2. Ensure that the cluster has been installed, including the HDFS, Yarn, Spark2x, and Kafka services.
3. Create a topic.

   {zkQuorum} indicates ZooKeeper cluster information in the IP address:Port number format.

   $KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper {zkQuorum}/kafka --replication-factor 1 --partitions 1 --topic {Topic}

4. Start the Producer of Kafka and send data to Kafka.

   {ClassPath} indicates the storage path of the project JAR package that is specified by the user. For details, see Compiling and Running the Application.

   java -cp $SPARK_HOME/conf:$SPARK_HOME/jars/*:$SPARK_HOME/jars/streamingClient010/*:{ClassPath} com.huawei.bigdata.spark.examples.KafkaProducer {brokerlist} {topic} {number of events produce every 0.02s}

   Example:

   java -cp /opt/client/Spark2x/spark/conf:/opt/StructuredStreamingState-1.0.jar:/opt/client/Spark2x/spark/jars/*:/opt/client/Spark2x/spark/jars/streamingClient010/* com.huawei.bigdata.spark.examples.KafkaProducer xxx.xxx.xxx.xxx:21005,xxx.xxx.xxx.xxx:21005,xxx.xxx.xxx.xxx:21005 mytopic 10

Development Guideline

1. Receive data from Kafka and generate the corresponding DataStreamReader.
2. Collect statistics by category.
3. Calculate the result and print it.

Packaging the Project

• Use the Maven tool provided by IDEA to pack the project and generate a JAR file.
• Upload the JAR package to any directory (for example, /opt) on the server where the Spark client is located.

Running Tasks

When running the example program, you need to specify <brokers>, <subscribe-type>, <topic>, and <checkpointLocation>, where <brokers> indicates the Kafka address for obtaining metadata (port 21005 is required). <subscribe-type> indicates the Kafka consumption mode, <topic> indicates the Kafka topic to be consumed, and <checkpointLocation> indicates the path for storing the checkpoint of the Spark task.

The path of the Spark Structured Streaming Kafka dependency package on the client is different from that of other dependency packages. For example, the path of other dependency packages is $SPARK_HOME/jars. Whereas the path of the Spark Streaming Structured Kafka dependency package is $SPARK_HOME/jars/streamingClient010. Therefore, when running an application, you need to add a configuration item to the spark-submit command to specify the path of the dependency package of Spark Streaming Kafka, for example, --jars $(files=($SPARK_HOME/jars/streamingClient010/*.jar); IFS=,; echo "${files[*]}")

Because the cluster is security mode, you need to add configuration items and modify the command parameters.

Go to the Spark client directory and run the following command to invoke the bin/spark-submit script to run the code (The class name and file name must be the same as those in the actual code. The following is only an example):

bin/spark-submit --master yarn --deploy-mode client --jars $(files=($SPARK_HOME/jars/streamingClient010/*.jar); IFS=,; echo "${files[*]}") --class com.huawei.bigdata.spark.examples.kafkaSessionization /opt/StructuredStreamingState-1.0.jar <brokers> <subscribe-type> <topic> <checkpointLocation>

When submitting a structured stream task, you need to run the --jars command to specify the path of the Kafka-related JAR file. For the current version, you need to cope the kafka-clientsjar file from the $SPARK_HOME/jars/streamingClient010 directory to the $SPARK_HOME/jars directory. Otherwise, the "class not found" error is reported.