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

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

Ensure that the cluster has been installed, including the HDFS, Yarn, Spark2x, and Kafka services.
Set the allow.everyone.if.no.acl.found parameter of Kafka Broker to true.
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}
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

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

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 example code is sparkuser, change the value to the prepared development user name.

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. (The file upload path must be the same as the path of the generated JAR file).

Running Tasks

When running the example program, you need to specify <brokers>, <subscribe-type>, <kafkaProtocol>, <kafkaService>, <kafkaDomain>, <topic>, and <checkpointLocation>, where <brokers> indicates the Kafka address for obtaining metadata (port 21007 is required). <subscribe-type> indicates the Kafka consumption mode, and <kafkaProtocol> indicates the secure access protocol (such as SASL_PLAINTEXT). <kafkaService> indicates the Kerberos service name (for example, kafka). <kafkaDomain> indicates the Kerberos domain name (for example, hadoop.<system domain name>). <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.

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

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

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;
};

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

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 --driver-java-options "-Djava.security.auth.login.config=<local path>/jaas.conf" --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> <kafkaProtocol> <kafkaService> <kafkaDomain> <topic> <checkpointLocation>
The configuration example is as follows:
```
--files ./jaas.conf,./user.keytab //Use --files to specify the jaas.conf and keytab 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.