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 clusters are installed, including HDFS, YARN, Spark2x and Kafka.
   • Set the allow.everyone.if.no.acl.found parameter of Kafka Broker to true.
   • 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 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.

Preparations

For clusters with the security mode enabled, 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 located.
• Use the Maven tool provided by IDEA to package the project and generate the JAR file. For details, see Commissioning a Spark Application in a Linux Environment.
  

  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. For example, /opt/female/user.keytab and /opt/female/krb5.conf.

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

Running the Task

When running the sample project, 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's authentication is in the 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 the following configuration to the $SPARK_HOME/conf/jaas-zk.conf file:

   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>

The configuration example is as follows:

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