Tutorial: Extracción, transformación y carga de datos mediante Azure Databricks

Artículo
03/08/2023

En este tutorial, realizará una operación de ETL (extracción, transformación y carga de datos) mediante Azure Databricks. Extraerá datos de Azure Data Lake Storage Gen2 a Azure Databricks, donde ejecutará transformaciones en ellos y, luego, cargará los datos transformados en Azure Synapse Analytics.

En los pasos de este tutorial se usa el conector Azure Synapse para Azure Databricks para transferir los datos a Azure Databricks. Este conector usa, a su vez, Azure Blob Storage como almacenamiento temporal para los datos transferidos entre un clúster de Azure Databricks y Azure Synapse.

La siguiente ilustración muestra el flujo de la aplicación:

Azure Databricks with Data Lake Store and Azure Synapse

En este tutorial se describen las tareas siguientes:

Cree un servicio de Azure Databricks.
Crear un clúster de Spark en Azure Databricks.
Crear un sistema de archivos en la cuenta de Data Lake Storage Gen2.
Cargar datos de ejemplo en la cuenta de Azure Data Lake Storage Gen2.
Crear una entidad de servicio.
Extraer datos de ejemplo de la cuenta de Azure Data Lake Storage Gen2.
Transformar datos en Azure Databricks.
Cargar datos en Azure Synapse.

Si no tiene una suscripción a Azure, cree una cuenta gratuita antes de empezar.

Nota:

Este tutorial no puede llevarse a cabo mediante una suscripción de evaluación gratuita de Azure. Si tiene una cuenta gratuita, vaya a su perfil y cambiar la suscripción a pago por uso. Para más información consulte el sitio de cuentas gratuitas de Azure. Después, quite el límite de gasto y solicite un aumento de la cuota para las vCPU de su región. Cuando crea su área de trabajo de Azure Databricks, puede seleccionar el plan de tarifa de la Trial (Premium - 14-Days Free DBUs) para que el área de trabajo acceda a las DBU Premium de Azure Databricks gratis durante 14 días.

Prerrequisitos

Complete estas tareas antes de comenzar este tutorial:

Cree una Azure Synapse, cree una regla de firewall de nivel de servidor y conéctese al servidor como administrador del servidor. Consulte Inicio rápido: Creación y consulta de un grupo de Synapse SQL mediante el Azure Portal.
Cree una clave maestra para la instancia de Azure Synapse. Consulte Crear la clave maestra de una base de datos.
Ha creado una cuenta de Azure Blob Storage y, dentro de ella, un contenedor. Además, ha recuperado la clave de acceso para acceder a la cuenta de almacenamiento. Consulte Inicio rápido: Carga, descarga y enumeración de blobs con Azure Portal.
Cree una cuenta de almacenamiento de Azure Data Lake Storage Gen2. Consulte Inicio rápido: Creación de una cuenta de almacenamiento de Azure Data Lake Storage Gen2.
Crear una entidad de servicio. Consulte Procedimiento: Uso del portal para crear una aplicación de Microsoft Entra ID (antes Azure Active Directory) y una entidad de servicio principal que pueda acceder a los recursos.

Hay un par de cosas que tendrá que hacer cuando realice los pasos de este artículo.
- Al realizar los pasos de la sección Asignación de la aplicación a un rol del artículo, asegúrese de asignar el rol Colaborador de datos de blobs de almacenamiento a la entidad de servicio en el ámbito de la cuenta de Data Lake Storage Gen2. Si asigna el rol al grupo de recursos o a la suscripción principales, recibirá errores relacionados con los permisos hasta que esas asignaciones de roles se propaguen a la cuenta de almacenamiento.
  
  Si prefiere usar una lista de control de acceso (ACL) para asociar la entidad de servicio con un archivo o un directorio específicos, consulte Control de acceso en Azure Data Lake Storage Gen2.
- Al realizar los pasos de la sección Obtención de valores para iniciar sesión del artículo, pegue los valores de identificador de inquilino, identificador de aplicación y secreto en un archivo de texto.
Inicie sesión en Azure Portal.

Recopilación de la información que necesita

Asegúrese de completar los requisitos previos de este tutorial.

Antes de comenzar, debe tener esta información:

✔️ Nombre de la base de datos, nombre del servidor de bases de datos, nombre de usuario y contraseña de Azure Synapse.

✔️ Clave de acceso de la cuenta de Blob Storage.

✔️ Nombre de la cuenta de almacenamiento de Data Lake Storage Gen2.

✔️ Identificador de inquilino de la suscripción.

✔️ El identificador de aplicación de la aplicación que registró con Microsoft Entra ID (anteriormente Azure Active Directory).

✔️ La clave de autenticación de la aplicación que registró con Microsoft Entra ID (anteriormente Azure Active Directory).

Creación de un servicio de Azure Databricks

En esta sección, va a crear un servicio de Azure Databricks con Azure Portal.

En el menú de Azure Portal, seleccione Crear un recurso.

A continuación, seleccione Análisis>Azure Databricks.

En Servicio de Azure Databricks, proporcione los valores siguientes para crear un servicio de Databricks:

Propiedad	Descripción
Workspace name (Nombre del área de trabajo)	Proporcione un nombre para el área de trabajo de Databricks.
Suscripción	En el cuadro desplegable, seleccione la suscripción de Azure.
Grupos de recursos	Especifique si desea crear un nuevo grupo de recursos o utilizar uno existente. Un grupo de recursos es un contenedor que almacena los recursos relacionados con una solución de Azure. Para más información, consulte Información general del grupo de recursos de Azure.
Location	Seleccione Oeste de EE. UU. 2. Para otras regiones disponibles, consulte Productos disponibles por región.
Plan de tarifa	Seleccione Estándar.

La cuenta tarda unos minutos en crearse. Para supervisar el estado de la operación, consulte la barra de progreso en la parte superior.
Seleccione Anclar al panel y, después, seleccione Crear.

Creación de un clúster de Spark en Azure Databricks

En Azure Portal, vaya al servicio de Databricks que ha creado y, después, seleccione Launch Workspace (Iniciar área de trabajo).
Se le redirigirá al portal de Azure Databricks. En el portal, seleccione Cluster.
En la página Nuevo clúster, proporcione los valores para crear un clúster.
Rellene los valores para los campos siguientes y acepte los valores predeterminados para el resto de campos:
- Escriba un nombre para el clúster.
- Asegúrese de que selecciona la casilla Terminar después de __ minutos de inactividad. Si no se usa el clúster, proporcione una duración (en minutos) para terminar el clúster.
- Seleccione Create cluster (Crear clúster). Después de ejecutar el clúster, puede asociarle cuadernos y ejecutar trabajos de Spark.

Creación de un sistema de archivos en la cuenta de Azure Data Lake Storage Gen2

En esta sección, creará un cuaderno en el área de trabajo de Azure Databricks y, a continuación, ejecutará fragmentos de código para configurar la cuenta de almacenamiento.

En Azure Portal, vaya al servicio de Azure Databricks que ha creado y, después, seleccione Launch Workspace (Iniciar área de trabajo).
A la izquierda, seleccione Workspace (Área de trabajo). En la lista desplegable Workspace (Área de trabajo), seleccione Create>Notebook (Crear > Cuaderno).
En el cuadro de diálogo Create Notebook (Crear cuaderno), escriba un nombre para el cuaderno. Seleccione Scala como lenguaje y, a continuación, seleccione el clúster de Spark que creó anteriormente.
Seleccione Crear.

El bloque de código siguiente establece las credenciales predeterminadas de la entidad de servicio de cualquier cuenta de ADLS Gen 2 a la que se accede en la sesión de Spark. El segundo bloque de código anexa el nombre de la cuenta a la configuración para especificar las credenciales para una cuenta específica de ADLS Gen 2. Copie y pegue cualquiera de estos bloques de código en la primera celda del cuaderno de Azure Databricks.

Configuración de la sesión

val appID = "<appID>"
val secret = "<secret>"
val tenantID = "<tenant-id>"

spark.conf.set("fs.azure.account.auth.type", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.id", "<appID>")
spark.conf.set("fs.azure.account.oauth2.client.secret", "<secret>")
spark.conf.set("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<tenant-id>/oauth2/token")
spark.conf.set("fs.azure.createRemoteFileSystemDuringInitialization", "true")

Configuración de la cuenta

val storageAccountName = "<storage-account-name>"
val appID = "<app-id>"
val secret = "<secret>"
val fileSystemName = "<file-system-name>"
val tenantID = "<tenant-id>"

spark.conf.set("fs.azure.account.auth.type." + storageAccountName + ".dfs.core.windows.net", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type." + storageAccountName + ".dfs.core.windows.net", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.id." + storageAccountName + ".dfs.core.windows.net", "" + appID + "")
spark.conf.set("fs.azure.account.oauth2.client.secret." + storageAccountName + ".dfs.core.windows.net", "" + secret + "")
spark.conf.set("fs.azure.account.oauth2.client.endpoint." + storageAccountName + ".dfs.core.windows.net", "https://login.microsoftonline.com/" + tenantID + "/oauth2/token")
spark.conf.set("fs.azure.createRemoteFileSystemDuringInitialization", "true")
dbutils.fs.ls("abfss://" + fileSystemName  + "@" + storageAccountName + ".dfs.core.windows.net/")
spark.conf.set("fs.azure.createRemoteFileSystemDuringInitialization", "false")

En este bloque de código, reemplace los valores de marcador de posición <app-id>, <secret>, <tenant-id> y <storage-account-name> por los valores que recopiló al completar los requisitos previos de este tutorial. Reemplace el valor del marcador de posición <file-system-name> por el nombre que desea dar al sistema de archivos.
- <app-id> y <secret> proceden de la aplicación que registró con Active Directory como parte de la creación de una entidad de servicio.
- <tenant-id> procede de su suscripción.
- <storage-account-name> es el nombre de la cuenta de almacenamiento de Azure Data Lake Storage Gen2.
Presione las teclas MAYÚS + ENTRAR para ejecutar el código de este bloque.

Ingesta de datos de ejemplo en la cuenta de Azure Data Lake Storage Gen2

Antes de empezar esta sección, debe completar lo siguientes requisitos previos:

Escriba el código siguiente en una celda del cuaderno:

%sh wget -P /tmp https://raw.githubusercontent.com/Azure/usql/master/Examples/Samples/Data/json/radiowebsite/small_radio_json.json

En la celda, presione MAYÚS + ENTRAR para ejecutar el código.

Ahora, en una nueva celda debajo de esta, escriba el siguiente código y reemplace los valores que aparecen entre paréntesis por los mismos valores que usó anteriormente:

dbutils.fs.cp("file:///tmp/small_radio_json.json", "abfss://" + fileSystemName + "@" + storageAccountName + ".dfs.core.windows.net/")

En la celda, presione MAYÚS + ENTRAR para ejecutar el código.

Extracción de datos de ejemplo de la cuenta de Azure Data Lake Storage Gen2

Ahora puede cargar el archivo JSON de ejemplo como trama de datos en Azure Databricks. Pegue el código siguiente en una nueva celda. Reemplace los marcadores de posición que se muestran entre paréntesis por sus valores.
```
val df = spark.read.json("abfss://" + fileSystemName + "@" + storageAccountName + ".dfs.core.windows.net/small_radio_json.json")
```
Presione las teclas MAYÚS + ENTRAR para ejecutar el código de este bloque.

Ejecute el código siguiente para ver el contenido de la trama de datos:

df.show()

Debe ver una salida similar al siguiente fragmento de código:

+---------------------+---------+---------+------+-------------+----------+---------+-------+--------------------+------+--------+-------------+---------+--------------------+------+-------------+------+
|               artist|     auth|firstName|gender|itemInSession|  lastName|   length|  level|            location|method|    page| registration|sessionId|                song|status|           ts|userId|
+---------------------+---------+---------+------+-------------+----------+---------+-------+--------------------+------+--------+-------------+---------+--------------------+------+-------------+------+
| El Arrebato         |Logged In| Annalyse|     F|            2|Montgomery|234.57914| free  |  Killeen-Temple, TX|   PUT|NextSong|1384448062332|     1879|Quiero Quererte Q...|   200|1409318650332|   309|
| Creedence Clearwa...|Logged In|   Dylann|     M|            9|    Thomas|340.87138| paid  |       Anchorage, AK|   PUT|NextSong|1400723739332|       10|        Born To Move|   200|1409318653332|    11|
| Gorillaz            |Logged In|     Liam|     M|           11|     Watts|246.17751| paid  |New York-Newark-J...|   PUT|NextSong|1406279422332|     2047|                DARE|   200|1409318685332|   201|
...
...

Ahora ha extraído los datos de Azure Data Lake Storage Gen2 en Azure Databricks.

Transformación de datos en Azure Databricks

Los datos de ejemplo sin procesar del archivo small_radio_json.json capturan la audiencia de una emisora de radio y presentan una variedad de columnas. En esta sección, va a transformar los datos para recuperar solo columnas específicas del conjunto de datos.

En primer lugar, recupere solo las columnas firstName, lastName, gender, location y level de la trama de datos que ha creado.

val specificColumnsDf = df.select("firstname", "lastname", "gender", "location", "level")
specificColumnsDf.show()

Recibirá una salida como la que se muestra en el fragmento de código siguiente:

+---------+----------+------+--------------------+-----+
|firstname|  lastname|gender|            location|level|
+---------+----------+------+--------------------+-----+
| Annalyse|Montgomery|     F|  Killeen-Temple, TX| free|
|   Dylann|    Thomas|     M|       Anchorage, AK| paid|
|     Liam|     Watts|     M|New York-Newark-J...| paid|
|     Tess|  Townsend|     F|Nashville-Davidso...| free|
|  Margaux|     Smith|     F|Atlanta-Sandy Spr...| free|
|     Alan|     Morse|     M|Chicago-Napervill...| paid|
|Gabriella|   Shelton|     F|San Jose-Sunnyval...| free|
|   Elijah|  Williams|     M|Detroit-Warren-De...| paid|
|  Margaux|     Smith|     F|Atlanta-Sandy Spr...| free|
|     Tess|  Townsend|     F|Nashville-Davidso...| free|
|     Alan|     Morse|     M|Chicago-Napervill...| paid|
|     Liam|     Watts|     M|New York-Newark-J...| paid|
|     Liam|     Watts|     M|New York-Newark-J...| paid|
|   Dylann|    Thomas|     M|       Anchorage, AK| paid|
|     Alan|     Morse|     M|Chicago-Napervill...| paid|
|   Elijah|  Williams|     M|Detroit-Warren-De...| paid|
|  Margaux|     Smith|     F|Atlanta-Sandy Spr...| free|
|     Alan|     Morse|     M|Chicago-Napervill...| paid|
|   Dylann|    Thomas|     M|       Anchorage, AK| paid|
|  Margaux|     Smith|     F|Atlanta-Sandy Spr...| free|
+---------+----------+------+--------------------+-----+

Puede transformar aún más estos datos para renombrar la columna level como subscription_type.

val renamedColumnsDF = specificColumnsDf.withColumnRenamed("level", "subscription_type")
renamedColumnsDF.show()

Recibirá una salida como la que se muestra en el fragmento de código siguiente.

+---------+----------+------+--------------------+-----------------+
|firstname|  lastname|gender|            location|subscription_type|
+---------+----------+------+--------------------+-----------------+
| Annalyse|Montgomery|     F|  Killeen-Temple, TX|             free|
|   Dylann|    Thomas|     M|       Anchorage, AK|             paid|
|     Liam|     Watts|     M|New York-Newark-J...|             paid|
|     Tess|  Townsend|     F|Nashville-Davidso...|             free|
|  Margaux|     Smith|     F|Atlanta-Sandy Spr...|             free|
|     Alan|     Morse|     M|Chicago-Napervill...|             paid|
|Gabriella|   Shelton|     F|San Jose-Sunnyval...|             free|
|   Elijah|  Williams|     M|Detroit-Warren-De...|             paid|
|  Margaux|     Smith|     F|Atlanta-Sandy Spr...|             free|
|     Tess|  Townsend|     F|Nashville-Davidso...|             free|
|     Alan|     Morse|     M|Chicago-Napervill...|             paid|
|     Liam|     Watts|     M|New York-Newark-J...|             paid|
|     Liam|     Watts|     M|New York-Newark-J...|             paid|
|   Dylann|    Thomas|     M|       Anchorage, AK|             paid|
|     Alan|     Morse|     M|Chicago-Napervill...|             paid|
|   Elijah|  Williams|     M|Detroit-Warren-De...|             paid|
|  Margaux|     Smith|     F|Atlanta-Sandy Spr...|             free|
|     Alan|     Morse|     M|Chicago-Napervill...|             paid|
|   Dylann|    Thomas|     M|       Anchorage, AK|             paid|
|  Margaux|     Smith|     F|Atlanta-Sandy Spr...|             free|
+---------+----------+------+--------------------+-----------------+

Carga de datos en Azure Synapse

En esta sección, cargará los datos transformados en Azure Synapse. Utilice el conector de Azure Synapse para Azure Databricks para cargar directamente una trama de datos como una tabla en un grupo de Synapse Spark.

Como se mencionó anteriormente, el conector de Azure Synapse usa Azure Blob Storage como almacenamiento temporal para cargar datos entre Azure Databricks y Azure Synapse. Por lo tanto, para comenzar, proporcione la configuración para conectarse a la cuenta de almacenamiento. Ya debe haber creado la cuenta como parte de los requisitos previos de este artículo.

Proporcione la configuración para acceder a la cuenta de Azure Storage desde Azure Databricks.

val blobStorage = "<blob-storage-account-name>.blob.core.windows.net"
val blobContainer = "<blob-container-name>"
val blobAccessKey =  "<access-key>"

Especifique una carpeta temporal que usará al mover datos entre Azure Databricks y Azure Synapse.
```
val tempDir = "wasbs://" + blobContainer + "@" + blobStorage +"/tempDirs"
```
Ejecute el siguiente fragmento de código para almacenar las claves de acceso de Azure Blob Storage en la configuración. Esta acción garantiza que no tiene que guardar la clave de acceso en el cuaderno en texto sin formato.
```
val acntInfo = "fs.azure.account.key."+ blobStorage
sc.hadoopConfiguration.set(acntInfo, blobAccessKey)
```

Proporcione los valores para conectarse a la instancia de Azure Synapse. Debe haber creado un servicio Azure Synapse Analytics como requisito previo. Use el nombre completo del servidor para dwServer. Por ejemplo, <servername>.database.windows.net.

//Azure Synapse related settings
val dwDatabase = "<database-name>"
val dwServer = "<database-server-name>"
val dwUser = "<user-name>"
val dwPass = "<password>"
val dwJdbcPort =  "1433"
val dwJdbcExtraOptions = "encrypt=true;trustServerCertificate=true;hostNameInCertificate=*.database.windows.net;loginTimeout=30;"
val sqlDwUrl = "jdbc:sqlserver://" + dwServer + ":" + dwJdbcPort + ";database=" + dwDatabase + ";user=" + dwUser+";password=" + dwPass + ";$dwJdbcExtraOptions"
val sqlDwUrlSmall = "jdbc:sqlserver://" + dwServer + ":" + dwJdbcPort + ";database=" + dwDatabase + ";user=" + dwUser+";password=" + dwPass

Ejecute el siguiente fragmento de código para cargar la trama de datos transformada, renamedColumnsDF, como una tabla en Azure Synapse. Este fragmento de código crea una tabla denominada SampleTable en SQL Database.
```
spark.conf.set(
    "spark.sql.parquet.writeLegacyFormat",
    "true")

renamedColumnsDF.write.format("com.databricks.spark.sqldw").option("url", sqlDwUrlSmall).option("dbtable", "SampleTable")       .option( "forward_spark_azure_storage_credentials","True").option("tempdir", tempDir).mode("overwrite").save()
```
Nota:

Este ejemplo utiliza la marca forward_spark_azure_storage_credentials, que hace que Azure Synapse acceda a los datos desde Blob Storage mediante una clave de acceso. Este es el único método admitido de autenticación.

Si Azure Blob Storage está restringido para seleccionar redes virtuales, Azure Synapse requiere Managed Service Identity en lugar de las claves de acceso. Esto provocará el error de solicitud no autorizada para realizar operación.
Conéctese a la instancia de SQL Database y compruebe que aparece la base de datos denominada SampleTable.
Ejecute una consulta select para comprobar el contenido de la tabla. La tabla debe tener los mismos datos que la trama de datos renamedColumnsDF.

Limpieza de recursos

Después de terminar el tutorial, puede finalizar el clúster. Desde el área de trabajo de Azure Databricks, seleccione Clusters (Clústeres) a la izquierda. Para que el clúster termine, en Acciones, apunte a los puntos suspensivos (...) y seleccione el icono Terminate (Finalizar).

Stop a Databricks cluster

Si no finaliza manualmente el clúster, este se detendrá automáticamente si seleccionó la casilla Finalizar después de __ minutos de inactividad al crear el clúster. En tal caso, el clúster se detiene automáticamente si ha estado inactivo durante el tiempo especificado.

Pasos siguientes

En este tutorial, ha aprendido a:

Creación de un servicio de Azure Databricks
Creación de un clúster de Spark en Azure Databricks
Creación de un cuaderno en Azure Databricks
Extracción de datos desde la cuenta de Data Lake Storage Gen2
Transformación de datos en Azure Databricks
Carga de datos en Azure Synapse