Desarrollo de aplicaciones de Spark

Por ejemplo, instale el cliente en el directorio /opt/client en el nodo de gestión activo.

Elija System > Permission > User. En el panel derecho, haga clic en Create. En la página mostrada, cree un usuario hombre-máquina, por ejemplo, sparkuser.

Agregue el grupo de usuarios supergroup y asocie el rol System_administrator.

Descargue el código fuente del proyecto Maven y los archivos de configuración del proyecto de ejemplo, y configure las herramientas de desarrollo relacionadas en su PC local. Para obtener más información, consulte Obtención de proyectos de muestra desde Huawei Mirros.

Seleccione una rama basada en la versión del clúster y descargue el proyecto de muestra de MRS requerido.

Por ejemplo, el proyecto de muestra adecuado para esta práctica es SparkHivetoHbase, que se puede obtener en https://github.com/huaweicloud/huaweicloud-mrs-example/tree/mrs-3.1.5/src/spark-examples/sparksecurity-examples/SparkHivetoHbaseJavaExample.

Después de configurar los parámetros Maven y SDK en el PC local, el proyecto de ejemplo carga automáticamente paquetes de dependencias relacionados. Para obtener más información, consulte Configuración e importación de proyectos de muestra.

Figura 1 Proyecto de muestra Spark Hive a HBase

La clase SparkHivetoHbase del proyecto de ejemplo utiliza Spark para invocar a las API de Hive para operar una tabla Hive, obtener el registro correspondiente de una tabla HBase basada en la clave, realizar operaciones en los dos registros de datos y actualizar los datos a la tabla HBase.

El fragmento de código es el siguiente:

...
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();
                }
            }
        }
    }
...

Para un clúster MRS con autenticación Kerberos activada, la aplicación debe realizar la autenticación de usuario en el servidor. En este proyecto de ejemplo, configure la información de autenticación en código. Establezca userPrincipal en el nombre de usuario para la autenticación y cambie userKeytabPath y krb5ConfPath a las rutas de archivos reales en el servidor de cliente.

En la ventana Maven, seleccione clean en Lifecycle para ejecutar el proceso de construcción de Maven. Seleccione package y obtenga el paquete JAR del directorio target.

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

Por ejemplo, el archivo JAR es SparkHivetoHbase-1.0.jar.

cd /opt/client

source bigdata_env

kinit sparkuser

beeline

En Hive Beeline, ejecute los siguientes comandos para crear una tabla e insertar datos:

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             |
+--------------+-----------------+

Salga de Hive Beeline, ejecute el comando spark-beeline y ejecute el siguiente comando para crear una tabla HBase:

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

Salga de Spark Beeline, ejecute el comando hbase shell para ir al HBase Shell y ejecute los siguientes comandos para insertar datos:

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

Cambie el valor de spark.yarn.security.credentials.hbase.enabled a 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)