Customizing a SparkStreaming Job

Obtaining the DIS Spark Streaming Demo

Obtain the dis-spark-streaming-X.X.X.zip package from https://dis-publish.obs-website.cn-north-1.myhuaweicloud.com/. Decompress the dis-spark-streaming-X.X.X.zip package to obtain the following directory that contains a Maven project sample:
dis-spark-streaming-demo: The dis-spark-streaming-demo directory contains a maven project sample.

Building a SparkStreaming Job

The following uses the IntelliJ IDEA community version as an example to describe how to build a SparkStreaming job. Make sure that the SparkStreaming job has been configured on the IDEA.

JDK 1.8+
Scala-sdk-2.11
Maven 3.3.*

Start IntelliJIDEA and choose File > Open. Select the dis-spark-streaming-demo directory and double-click pom.xml.
When the following dialog box is displayed, select Open as Project.
Click New Window to open the project in a new window.
In the displayed IDEA window, choose File > Settings.
Enter maven in the search box and find the maven configuration. Ensure that Maven home directory (maven installation path), User settings file (settings.xml file location), and Local repository (local repository address) are correctly configured.

If configurations are incorrect, modify them. Otherwise, the SDK installed in Step 2 cannot be found.
Open the DISSparkStreamingExample file. If the IDEA displays the message "No Scala SDK in module", click Setup Scala SDK next to it and select one from the Scala SDK list. If no Scala SDK is available, create one and associate it with the Scala path. Version 2.11 can be selected.
Right-click pom.xml and choose Maven > Reimport from the shortcut menu to import the maven dependency library again.
If there is no error in the DISSparkStreamingExample file opened by IDEA, the development environment is configured successfully. The logic of this file is to read the data in DIS streams and count the occurrence times of each word.
1. DISSparkStreamingExample: an example of using the Assign mode. It does not have the capability to start from the last stop. The SDK construction method is as follows:
```
ConsumerStrategies.Assign[String, String](streamName, params, startingOffsets)
```
- streamName: a DIS stream name.
- params: a set of parameter maps. It includes at least endpoint (DIS gateway address), region (region where DIS resides), ak (user AK), sk (user SK), and projectId (user project ID).
- startingOffsets: start position for reading DIS data. LATEST indicates read from the latest data. EARLIEST indicates reading from the oldest data. If the exact start position of each partition needs to be specified, the value can be a JSON character string. For example, {"0" :23, "1" :-1, "2" :-2}, which indicates that the start position of partition 0 is 23, the first partition starts from the location of the latest data, and the second partition starts from the position of the earliest data. If there is no specified location for a partition, the system starts from the latest data position by default.
1. DISSparkStreamingSubscribeExample: an example of using the Subscribe mode. It has the capability to start from the last stop. The SDK construction method is as follows:
```
ConsumerStrategies.Subscribe[String, String](Array(streamName), params)
```
- streamName: a DIS stream name.
- params: a set of parameter maps. It includes at least endpoint (DIS gateway address), region (region where DIS resides), ak (user AK), sk (user SK), projectId (user project ID), and group.id (app name, indicating a consumer group). It can also include auto.offset.reset. The meaning of this parameter is the same as that of startingOffsets in Assign mode. If enable.auto.commit is set to true, the system automatically submits the offset every 5000 ms (which can be modified by setting auto.commit.interval.ms). If the parameter is set to false, the system does not automatically submit the offset. You can manually invoke commitAsync to submit the offset. For details, see the following part in the sample code:
```
stream.foreachRDD { rdd =>
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      // commit offset to DIS async.
      stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}
```

Testing a SparkStreming Job

This section describes how to test a SparkStreming job in the local IDE to understand the basic usage of the SDK. In a real-world scenario, the SparkStreming job needs to run on a Spark cluster. After the test is complete, you can create clusters (such as MRS clusters) and submit a job for verification.

Use the account to log in to the DIS console.
Click in the upper left corner of the page and select a region and project.
Create a DIS stream by referring to Step 1: Creating a DIS Stream and continuously upload data to the newly created DIS stream. In this example, the content to be uploaded is hello world.
Open the pom.xml file, press Ctrl+/ to comment out the <scope>provided</scope> row, and save the setting.
Right-click pom.xml and choose Maven > Reimport from the shortcut menu to import the dependency package again.
Right-click in the DISSparkStreamingExample file and choose Create 'DISSparkStreamingExample' from the shortcut menu.
On the displayed configuration page, enter -Dspark.master=local[*] in VM options, indicating that a Spark job runs in local mode. Enter running parameters in Program arguments in the following format:
```
DIS gateway address Region name AK SK ProjectID Stream name Start position Streaming batch time
 
https://dis.${region}.cloud.com ${region} YOU_AK YOU_SK YOU_PROJECTID YOU_STREAM_NAME latest 10
```
The parameter sequence and meaning are available in the sample code. For details, see the following information:

The final configurations of the IDEA are shown in the following figure. After confirming that all configurations are correct, click OK to close the window.
Right-click in the DISSparkStreamingExample file and choose Run 'DISSparkStreamingExample' from the shortcut menu to start the job.

During the startup, an error message "hadoop binary path" is displayed, which can be ignored.

18/08/28 10:26:10 ERROR Shell: Failed to locate the winutils binary in the hadoop binary path
java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.

If there are no other errors, the job runs a batch at a fixed interval, reads data from the batch, and exports the read result. The following is an example:
```
-------------------------------------------
Time: 1535423650000 ms
-------------------------------------------
(hello,30)
(world.,30)
```
After verifying that the job can run locally without error, remove the comment tag from the <scope>provided</scope> row in pom.xml to prevent the Spark dependency from being packaged. Then stop the data upload program.