Overview

Scenarios

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 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 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.

Configuration Operations Before Running

In security mode, the Spark Core sample code needs to read two files (user.keytab and krb5.conf). The user.keytab and krb5.conf files are authentication files in the security mode. Download the authentication credentials of the user principal on the FusionInsight Manager page. The user in the sample code is sparkuser, change the value to the prepared development user name.

Packaging the Project

• Upload the user.keytab and krb5.conf files to the server where the client is installed.
• Use the Maven tool provided by IDEA to pack the project and generate a JAR file. For details, see Compiling and Running the Application.
  

  Before compilation and packaging, change the paths of the user.keytab and krb5.conf files in the sample code to the actual paths on the client server where the files are located. Example: /opt/female/user.keytab or /opt/female/krb5.conf.

• Upload the JAR package to any directory (for example, /opt) on the server where the Spark client is located.

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, and <kafkaProtocol> indicates the secure access protocol (for example, SASL_PLAINTEXT). <kafkaService> indicates the Kerberos service name (for example, kafka), and <kafkaDomain> indicates the Kerberos domain name (for example, hadoop.<system domain name>).

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

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

1. Add configuration items to $SPARK_HOME/conf/jaas.conf:

   KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=false
   useTicketCache=true
   debug=false;
   };

2. Add configuration items to $SPARK_HOME/conf/jaas-zk.conf

   KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="./user.keytab"
   principal="sparkuser@<System domain name>"
   useTicketCache=false
   storeKey=true
   debug=true;
   };

3. Use --files and relative path to submit the keytab file to ensure that the keytab file is loaded to the container of the executor.

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 --files <local Path>/jaas.conf,<local path>/user.keytab --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> <kafkaProtocol> <kafkaService> <kafkaDomain>

Go to the Spark client directory and run the following command to invoke the bin/spark-submit script to run the code:

--files ./jaas.conf,./user.keytab //Use --files to specify the jaas.conf and keytab files.