Trabalhando com o catálogo do Apache Hudi

O tópico descreve o procedimento para executar um aplicativo Spark que alimenta dados em um Catálogo do Apache Hudi

Crie um armazenamento com o catálogo Apache Hudi para armazenar dados usados no aplicativo Spark. Para criar armazenamento com o catálogo do Apache Hudi, consulte Incluindo um par de catálogo de armazenamento.
Associe o armazenamento ao mecanismo do Spark externo Para obter mais informações, consulte Associando um catálogo a um mecanismo.
Crie o Cloud Object Storage (COS) para armazenar o aplicativo Spark. Para criar Cloud Object Storage e um depósito, consulte Criando um depósito de armazenamento.
Registre o Cloud Object Storage em watsonx.data. Para obter mais informações, consulte Incluindo um par de catálogos de armazenamento

Salve o aplicativo Spark a seguir (arquivoPython ) em sua máquina local. Aqui, hudi_demo.py..

O aplicativo Spark Python demonstra a funcionalidade a seguir:

Ele cria um banco de dados dentro do catálogo Apache Hudi (que você criou para armazenar dados). Aqui, <database_name>..
Ele cria uma tabela no banco de dados <database_name>, chamada <table_name>.
Ele insere dados na <table_name> e faz a operação de consulta SELECT.
Ele elimina a tabela e o esquema após o uso

    from pyspark.sql import SparkSession

    def init_spark():
        spark = SparkSession.builder \
            .appName("CreateHudiTableInCOS") \
            .enableHiveSupport() \
            .getOrCreate()
        return spark

    def main():

        try:
            spark = init_spark()
            spark.sql("show databases").show()
            spark.sql("create database if not exists spark_catalog.<database_name> LOCATION 's3a://<data_storage_name>/'").show()
            spark.sql("create table if not exists spark_catalog.<database_name>.<table_name> (id bigint, name string, location string) USING HUDI OPTIONS ('primaryKey' 'id', hoodie.write.markers.type= 'direct', hoodie.embed.timeline.server= 'false')").show()
            spark.sql("insert into <database_name>.<table_name> VALUES (1, 'Sam','Kochi'), (2, 'Tom','Bangalore'), (3, 'Bob','Chennai'), (4, 'Alex','Bangalore')").show()
            spark.sql("select * from spark_catalog.<database_name>.<table_name>").show()
            spark.sql("drop table spark_catalog.<database_name>.<table_name>").show()
            spark.sql("drop schema spark_catalog.<database_name> CASCADE").show()

        finally:
            spark.stop()

    if __name__ == '__main__':
        main()

Faça upload do aplicativo Spark para o COS, consulte Fazendo upload de dados.

Para enviar o aplicativo Spark com dados que residem no Cloud Object Storage, especifique os valores de parâmetro e execute o comando curl a seguir

curl --request POST \
--url https://api.<region>.ae.cloud.ibm.com/v3/analytics_engines/<iae-instance-guid>/spark_applications \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--data '{
		"conf": {
        "spark.serializer" : "org.apache.spark.serializer.KryoSerializer",
        "spark.hadoop.fs.s3a.bucket.<data_storage_name>.endpoint" : "<your_data_bucket_direct_endpoint>",
        "spark.hadoop.fs.s3a.bucket.<data_storage_name>.access.key" : "<data_bucket_access_key>",
        "spark.hadoop.fs.s3a.bucket.<data_storage_name>.secret.key" : "<data_bucket_secret_key>",
        "spark.hadoop.fs.s3a.path.style.access" : "true",
        "spark.hadoop.fs.s3a.impl" : "org.apache.hadoop.fs.s3a.S3AFileSystem",
        "spark.hive.metastore.uris" : "<metastore URL>",
        "spark.hive.metastore.use.SSL" : "true",
        "spark.hive.metastore.truststore.path" : "file:///opt/ibm/jdk/lib/security/cacerts",
        "spark.hive.metastore.truststore.password" : "changeit",
        "spark.hive.metastore.truststore.type" : "JKS",
        "spark.hive.metastore.client.auth.mode" : "PLAIN",
        "spark.hive.metastore.client.plain.username" : "ibmlhapikey",
        "spark.hive.metastore.client.plain.password" : "<wxd_api_key>",
        "spark.driver.extraJavaOptions" : "-Dcom.sun.jndi.ldap.object.disableEndpointIdentification=true -Djdk.tls.trustNameService=true",
        "spark.executor.extraJavaOptions" : "-Dcom.sun.jndi.ldap.object.disableEndpointIdentification=true -Djdk.tls.trustNameService=true",
        "spark.hadoop.hive.metastore.schema.verification" : "false",
        "spark.hadoop.hive.metastore.schema.verification.record.version" : "false",
    "spark.sql.extensions": "org.apache.spark.sql.hudi.HoodieSparkSessionExtension",
       "spark.kryo.registrator": "org.apache.spark.HoodieSparkKryoRegistrar",
       "spark.sql.catalog.spark_catalog.type": "hudi",
        "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.hudi.catalog.HoodieCatalog"

		},
		"application": "s3a://<data_storage_name>/hudi_demo.py"
	}

Valores de parâmetro:

<region>: a região na qual você provisiona a instância do mecanismo do Analytics.
<iae-instance-guid>: o GUID da instância do mecanismo do Analytics. Para obter isso, consulte Recuperando detalhes de uma instância sem servidor
<token>: O token de portador. Para obter mais informações sobre como gerar o token, consulte token do IAM.
<your_data_bucket_direct_endpoint> : o terminal direto para acessar o depósito de dados Exemplo, s3.us-south.cloud-object-storage.appdomain.cloud para um depósito de armazenamento de objeto de nuvem na região us-south. Para obter mais informações, consulte Credenciais de serviço
<data_bucket_access_key> : a chave de acesso para o armazenamento do Cloud Object (armazenamento de dados). Para obter mais informações, consulte Criar credenciais do HMAC usando a CLI.
<data_bucket_secret_key> : a chave secreta para o armazenamento do Cloud Object (armazenamento de dados). Para obter mais informações, consulte Criar credenciais do HMAC usando a CLI.
<metastore URL>: O site URL do catálogo. Para obter mais informações, consulte Obtendo o ponto de extremidade MDS.
<wxd_api_key>: para gerar a chave de API, efetue login no console watsonx.data e navegue para Perfil> Perfil e configurações> Chaves de API e gere uma nova chave de API.

Após enviar o aplicativo Spark, você receberá uma mensagem de confirmação com o ID do aplicativo e a versão do Spark. Guarde para referência.
Efetue login no cluster watsonx.data, acesse a página Detalhes do mecanismo. Na guia Aplicativos, use o ID do aplicativo para listar o aplicativo e controlar os estágios Para obter mais informações, consulte Visualizar e gerenciar aplicativos