Help Center/ MapReduce Service/ Developer Guide (LTS)/ Spark2x Development Guide (Normal Mode)/ Developing the Project/ Structured Streaming Stream-Stream Join/ Overview
Updated on 2022-11-18 GMT+08:00
View PDF

Overview

Scenario

An advertisement service involves advertisement request events, advertisement display events, and advertisement click events. The advertiser needs to collect statistics on valid advertisement displays and advertisement click data.

Known conditions are as follows:

  1. Each time a user requests an advertisement, an advertisement request event is generated and saved to the adRequest topic of Kafka.
  2. After an advertisement is requested, the advertisement may be displayed for multiple times. Each time the advertisement is displayed, an advertisement display event is generated and saved to the adShow topic of Kafka.
  3. Each advertisement may be clicked for multiple times. Each time it is clicked, an advertisement click event is generated and saved to the adClick topic of Kafka.
  4. For advertisement display:
    1. If the duration from the time when a request is generated to the time when the advertisement is displayed exceeds A minutes, the display is invalid.
    2. Advertisement display events generated during A minutes are valid events.
  5. For advertisement click:
    1. If the duration from the display event to the click event exceeds B minutes, the click is invalid.
    2. If there are multiple click events within B minutes, only the first click event is valid.

The simple data structure in this scenario is as follows:

  • Advertisement request event

    Data structure: adID^reqTime

  • Advertisement display event

    Data structure: adID^showID^showTime

  • Advertisement click event

    Data structure: adID^showID^clickTime

Data association relationships are as follows:

  • The advertisement request event is associated with the advertisement display event through the adID.
  • The advertisement display event is associated with the advertisement click event through the adID and showID.

Data requirements:

  • The delay from the time when data is generated to the time when the data reaches the stream processing engine does not exceed two hours.
  • The time for advertisement request events, advertisement display events, and advertisement click events reach the stream processing engine may not be in sequence or be aligned with each other.

Data Planning

  1. Enable users with the Kafka permission to generate simulation data in Kafka.

    java -cp $SPARK_HOME/conf:$SPARK_HOME/jars/*:$SPARK_HOME/jars/streamingClient010/*:{ClassPath} com.huawei.bigdata.spark.examples.KafkaADEventProducer {BrokerList} {timeOfProduceReqEvent} {eventTimeBeforeCurrentTime} {reqTopic} {reqEventCount} {showTopic} {showEventMaxDelay} {clickTopic} {clickEventMaxDelay}

    • Ensure that the clusters are installed, including the HDFS, Yarn, Spark2x, and Kafka.
    • Set the allow.everyone.if.no.acl.found parameter of Kafka Broker to true.
    • Start Kafka Producer and enable it to send data to Kafka.
    • {ClassPath} indicates the path for storing the JAR package of the project. The path is specified by users. For details, see the step of exporting JAR packages in section Compiling and Running the Application.

    Command example:

    java -cp /opt/client/Spark2x/spark/conf:/opt/StructuredStreamingADScalaExample-1.0.jar:/opt/client/Spark2x/spark/jars/*:/opt/client/Spark2x/spark/jars/streamingClient010/* com.huawei.bigdata.spark.examples.KafkaADEventProducer 10.132.190.170:21005,10.132.190.165:21005 2h 1h req 10000000 show 5m click 5m

    This command creates three topics on Kafka: req, show, and click. Ten million data records of request events are generated in two hours. The time range of the request events is from one hour ahead of the current time to the current time, and up to five display events are randomly generated for each request event. The time range of the display events is from the request event time to five minutes after the request event time. Up to five click events are randomly generated for each display event. The time range of the click events is from the display event time to five minutes after the display event time.

Development Guideline

  1. Use structured streaming to receive data from Kafka and generate request flows, display flows, and click flows.
  2. Perform join query of data in request flows, display flows, and click flows.
  3. Write the statistics result to Kafka.
  4. Monitor the flow processing task status in the application.

Packaging the Project

  • Use the Maven tool provided by IDEA to pack the project and generate a JAR file. For details, see Compiling and Running the Application.
  • Upload the JAR package to any directory (for example, /opt) on the server where the Spark client is located.
  • Upload the user.keytab and krb5.conf files to the server where the client is installed.

Running Tasks

When running the sample program, you need to specify <kafkaBootstrapServers>, <maxEventDelay>, <reqTopic>, <showTopic>, <maxShowDelay>, <clickTopic>, <maxClickDelay>, <triggerInterver>, <checkpointDir>, <kafkaProtocol>, <kafkaService>, and <kafkaDomain>. <kafkaBootstrapServers> indicates the Kafka address for obtaining metadata (port 21007 is required), and <maxEventDelay> indicates the maximum delay from data generation to stream processing. <reqTopic> indicates the topic name of the request event. <showTopic> indicates the topic name of the display event. <maxShowDelay> indicates the maximum delay for displaying the event, and <clickTopic> indicates the topic name of the click event. <maxClickDelay> indicates the maximum delay time of a valid click event. <triggerInterver> indicates the interval for triggering a stream processing task. <checkpointDir> indicates the path for storing the checkpoint file.

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 Structured Streaming 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[*]}")

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.

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[*]}") --conf "spark.sql.streaming.statefulOperator.checkCorrectness.enabled=false" --class com.huawei.bigdata.spark.examples.KafkaADCount /opt/StructuredStreamingADScalaExample-1.0.jar <kafkaBootstrapServers> <maxEventDelay> <reqTopic> <showTopic> <maxShowDelay> <clickTopic> <maxClickDelay> <triggerInterver> <checkpointDir>