Development Plan

Overview

Suppose an advertising service has three events: advertisement request, advertisement display, and advertisement click. An advertiser requires real-time statistics on effective advertisement display and 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

  Data structure: adID^reqTime

• Advertisement display

  Data structure: adID^showID^showTime

• Advertisement click

  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 data generated should reach the stream processing engine within two hours.
• The sequence and alignment of advertisement request, display, and click events reaching the stream processing engine may vary.

Preparing Data

1. Generate simulated data in Kafka (the Kafka permission is required).
   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 cluster is installed with 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 storage path of the project JAR package that is specified by the user. For details, see Commissioning a Spark Application in a Linux Environment.

   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 sets up three Kafka topics: req, show, and click. Within two hours, ten million request event data records are generated. The request events' time range is from one hour ahead of the current time to the current time, and each request event can have up to five randomly generated display events. The display events' time range is from the request event time to five minutes after the request event time. Each display event can have up to five randomly generated click events, with the click events' time range being from the display event time to five minutes after the display event time.

Development Guidelines

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 Commissioning a Spark Application in a Linux Environment.
• 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 located.

Running the Task

When running the sample project, you need to specify <kafkaBootstrapServers>, <maxEventDelay>, <reqTopic>, <showTopic>, <maxShowDelay>, <clickTopic>, <maxClickDelay>, <triggerInterver>, and <checkpointDir. <checkpointDir> indicates the path for storing the checkpoint file.

• <kafkaBootstrapServers> indicates the Kafka address for obtaining metadata.
• <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 effectively displaying the event.
• <clickTopic> indicates the topic name of the click event.
• <maxClickDelay> indicates the maximum delay of a valid click event.
• <triggerInterver> indicates the interval for triggering a stream processing task.
• <checkpointDir> indicates the path for storing the checkpoint file, which can be a local path or an HDFS path.

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>