Help Center/ MapReduce Service/ Developer Guide (LTS)/ Spark2x Development Guide (Security Mode)/ Developing Spark Applications/ Sample Project for Spark Structured Streaming Status Operations/ Development Plan
Updated on 2024-08-10 GMT+08:00
View PDF
Share

Development Plan

Overview

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.

Preparing Data

  1. Generate simulated data in Kafka (the Kafka permission is required).
  2. Ensure that the cluster has been installed, including the HDFS, Yarn, Spark2x, and Kafka services.
  3. Set the allow.everyone.if.no.acl.found parameter of Kafka Broker to true.
  4. 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}

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

    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}

    The following is an 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 Guidelines

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

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

  • 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.
  • 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 the Task

When running the sample project, 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'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.

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