Spark Application Development

For example, install the client in the /opt/client directory on the active management node.

Choose System > Permission > User. In the right pane, click Create. On the displayed page, create a human-machine user, for example, sparkuser.

Add the supergroup user group and associate with the System_administrator role.

Download the Maven project source code and configuration files of the sample project, and configure related development tools on your local PC. For details, see Obtaining Sample Projects from Huawei Mirrors.

Select a branch based on the cluster version and download the required MRS sample project.

For example, the sample project suitable for this practice is SparkHivetoHbase, which can be obtained at https://github.com/huaweicloud/huaweicloud-mrs-example/tree/mrs-3.1.5/src/spark-examples/sparksecurity-examples/SparkHivetoHbaseJavaExample.

After you configure Maven and SDK parameters on the local PC, the sample project automatically loads related dependency packages. For details, see Configuring and Importing Sample Projects.

Figure 1 Spark Hive to HBase sample project

The SparkHivetoHbase class in the sample project uses Spark to call Hive APIs to operate a Hive table, obtain the corresponding record from an HBase table based on the key, perform operations on the two data records, and update the data to the HBase table.

The code snippet is as follows:

...
public class SparkHivetoHbase {
    public static void main(String[] args) throws Exception {
        String userPrincipal = "sparkuser";     //Specifies the cluster user information and keytab file address used for authentication.
        String userKeytabPath = "/opt/client/user.keytab";
        String krb5ConfPath = "/opt/client/krb5.conf";
        Configuration hadoopConf = new Configuration();
        LoginUtil.login(userPrincipal, userKeytabPath, krb5ConfPath, hadoopConf);
        //Calls the Spark API to obtain table data.
        SparkConf conf = new SparkConf().setAppName("SparkHivetoHbase");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        HiveContext sqlContext = new org.apache.spark.sql.hive.HiveContext(jsc);
        Dataset<Row> dataFrame = sqlContext.sql("select name, account from person");
        //Traverses partitions in the Hive table and updates the partitions to the HBase table.
                dataFrame
                .toJavaRDD()
                .foreachPartition(
                        new VoidFunction<Iterator<Row>>() {
                            public void call(Iterator<Row> iterator) throws Exception {
                                hBaseWriter(iterator);
                            }
                        });
        jsc.stop();
    }
    //Updates records in the HBase table on the executor.
    private static void hBaseWriter(Iterator<Row> iterator) throws IOException {
        //Reads the HBase table.
        String tableName = "table2";
        String columnFamily = "cf";
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf(tableName));
        try {
            connection = ConnectionFactory.createConnection(conf);
            table = connection.getTable(TableName.valueOf(tableName));
            List<Row> table1List = new ArrayList<Row>();
            List<Get> rowList = new ArrayList<Get>();
            while (iterator.hasNext()) {
                Row item = iterator.next();
                Get get = new Get(item.getString(0).getBytes());
                table1List.add(item);
                rowList.add(get);
            }
            //Obtains the records in the HBase table.
            Result[] resultDataBuffer = table.get(rowList);
            //Modifies records in the HBase table.
            List<Put> putList = new ArrayList<Put>();
            for (int i = 0; i < resultDataBuffer.length; i++) {
                Result resultData = resultDataBuffer[i];
                if (!resultData.isEmpty()) {
                    int hiveValue = table1List.get(i).getInt(1);
                    String hbaseValue = Bytes.toString(resultData.getValue(columnFamily.getBytes(), "cid".getBytes()));
                    Put put = new Put(table1List.get(i).getString(0).getBytes());
                    //Calculates the result.
                    int resultValue = hiveValue + Integer.valueOf(hbaseValue);
                    put.addColumn(
                            Bytes.toBytes(columnFamily),
                            Bytes.toBytes("cid"),
                            Bytes.toBytes(String.valueOf(resultValue)));
                    putList.add(put);
                }
            }
            if (putList.size() > 0) {
                table.put(putList);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    //Closes the HBase connection.
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
...

For an MRS cluster with Kerberos authentication enabled, the application needs to perform user authentication on the server. In this sample project, configure authentication information in code. Set userPrincipal to the username for authentication and change userKeytabPath and krb5ConfPath to the actual file paths on the client server.

In the Maven window, select clean from Lifecycle to execute the Maven building process. Select package and obtain the JAR package from the target directory.

[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  02:36 min
[INFO] Finished at: 2023-06-12T20:46:24+08:00
[INFO] ------------------------------------------------------------------------

For example, the JAR file is SparkHivetoHbase-1.0.jar.

cd /opt/client

source bigdata_env

kinit sparkuser

beeline

In the Hive Beeline, run the following commands to create a table and insert data:

create table person ( name STRING, account INT ) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' ESCAPED BY '\\' STORED AS TEXTFILE;

insert into table person(name,account) values("1","100");

select * from person;

+--------------+-----------------+
| person.name  | person.account  |
+--------------+-----------------+
| 1            | 100             |
+--------------+-----------------+

Exit the Hive Beeline, run the spark-beeline command, and run the following command to create an HBase table:

create table table2 ( key string, cid string ) using org.apache.spark.sql.hbase.HBaseSource options( hbaseTableName "table2", keyCols "key", colsMapping "cid=cf.cid" );

Exit the Spark Beeline, run the hbase shell command to go to the HBase Shell, and run the following commands to insert data:

put 'table2', '1', 'cf:cid', '1000'

scan 'table2'

ROW                                                 COLUMN+CELL                                                                                                                                           
 1                                                 column=cf:cid, timestamp=2023-06-12T21:12:50.711, value=1000                                                                                           
1 row(s)

cd /opt/client

source bigdata_env

cd Spark2x/spark

vi conf/spark-defaults.conf

Change the value of spark.yarn.security.credentials.hbase.enabled to true.

bin/spark-submit --class com.huawei.bigdata.spark.examples.SparkHivetoHbase --master yarn --deploy-mode client /opt/client/sparkdemo/SparkHivetoHbase-1.0.jar

cd /opt/client

source bigdata_env

hbase shell

scan 'table2'

ROW                                                 COLUMN+CELL                                                                                                                                           
 1                                                 column=cf:cid, timestamp=2023-06-12T21:22:50.711, value=1100                                                                                           
1 row(s)