Usar o Delta Lake no Azure HDInsight em AKS com o cluster Apache Spark™ (Prévia)

Artigo
09/20/2024

Importante

O Azure HDInsight no AKS foi desativado em 31 de janeiro de 2025. Saiba mais com este anúncio.

Você precisa migrar suas cargas de trabalho para Microsoft Fabric ou um produto equivalente do Azure para evitar o encerramento abrupto de suas cargas de trabalho.

Importante

Esta funcionalidade está atualmente em pré-visualização. Os Termos de Utilização Suplementares para Pré-visualizações do Microsoft Azure incluem mais termos legais que se aplicam às funcionalidades do Azure que estão em versão beta, em pré-visualização ou ainda não disponíveis para o público em geral. Para obter informações sobre essa visualização específica, consulte Azure HDInsight no AKS informações de visualização. Para perguntas ou sugestões de funcionalidades, envie uma solicitação no AskHDInsight com os detalhes e siga-nos para obter mais atualizações sobre a Comunidade do Azure HDInsight .

o Azure HDInsight no AKS é um serviço gerenciado baseado em nuvem para análise de big data que ajuda as organizações a processar grandes quantidades de dados. Este tutorial mostra como usar o Delta Lake no Azure HDInsight no AKS com cluster Apache Spark™.

Pré-requisito

Criar um cluster Apache Spark™ no Azure HDInsight no AKS
Execute o cenário Delta Lake no Jupyter Notebook. Crie um bloco de anotações Jupyter e selecione "Spark" ao criar um bloco de anotações, já que o exemplo a seguir está no Scala.

Cenário

Leia NYC Taxi Parquet Formato de dados - Lista de arquivos Parquet URLs são fornecidos pelo NYC Taxi & Limousine Commission.
Para cada url (arquivo) execute alguma transformação e armazene no formato Delta.
Calcule a distância média, o custo médio por milha e o custo médio da Tabela Delta usando carga incremental.
Armazene o valor calculado da Etapa #3 no formato Delta na pasta de saída da KPI.
Crie uma Tabela Delta na pasta de saída do Formato Delta (com atualização automática).
A pasta de saída KPI tem várias versões da distância média e do custo médio por milha para uma viagem.

Fornecer configurações necessárias para o lago delta

Delta Lake com matriz de compatibilidade com o Apache Spark - Delta Lake, altere a versão do Delta Lake com base na versão do Apache Spark.

%%configure -f
{ "conf": {"spark.jars.packages": "io.delta:delta-core_2.12:1.0.1,net.andreinc:mockneat:0.4.8",
"spark.sql.extensions":"io.delta.sql.DeltaSparkSessionExtension",
"spark.sql.catalog.spark_catalog":"org.apache.spark.sql.delta.catalog.DeltaCatalog"
}
  }

Listar o arquivo de dados

Observação

Estas URLs de ficheiro pertencem à Comissão de Táxis e Limousines de NYC &.

import java.io.File
import java.net.URL
import org.apache.commons.io.FileUtils
import org.apache.hadoop.fs._
    
// data file object is being used for future reference in order to read parquet files from HDFS
case class DataFile(name:String, downloadURL:String, hdfsPath:String)
    
// get Hadoop file system
val fs:FileSystem = FileSystem.get(spark.sparkContext.hadoopConfiguration)
    
val fileUrls= List(
"https://d37ci6vzurychx.cloudfront.net/trip-data/fhvhv_tripdata_2022-01.parquet"
    )
    
// Add a file to be downloaded with this Spark job on every node.
        val listOfDataFile = fileUrls.map(url=>{
        val urlPath=url.split("/") 
        val fileName = urlPath(urlPath.size-1)
        val urlSaveFilePath = s"/tmp/${fileName}"
        val hdfsSaveFilePath = s"/tmp/${fileName}"
        val file = new File(urlSaveFilePath)
        FileUtils.copyURLToFile(new URL(url), file)
        // copy local file to HDFS /tmp/${fileName}
        // use FileSystem.copyFromLocalFile(boolean delSrc, boolean overwrite, Path src, Path dst)
        fs.copyFromLocalFile(true,true,new org.apache.hadoop.fs.Path(urlSaveFilePath),new org.apache.hadoop.fs.Path(hdfsSaveFilePath))
        DataFile(urlPath(urlPath.size-1),url, hdfsSaveFilePath)
})

Criar diretório de saída

O local onde desejas criar a saída em formato delta; altera a variável transformDeltaOutputPath e avgDeltaOutputKPIPath, se necessário.

avgDeltaOutputKPIPath - para armazenar KPI médio em formato delta
transformDeltaOutputPath - armazene a saída transformada no formato delta

import org.apache.hadoop.fs._

// this is used to store source data being transformed and stored delta format
val transformDeltaOutputPath = "/nyctaxideltadata/transform"
// this is used to store Average KPI data in delta format
val avgDeltaOutputKPIPath = "/nyctaxideltadata/avgkpi"
// this is used for POWER BI reporting to show Month on Month change in KPI (not in delta format)
val avgMoMKPIChangePath = "/nyctaxideltadata/avgMoMKPIChangePath"

// create directory/folder if not exist
def createDirectory(dataSourcePath: String) = {
    val fs:FileSystem = FileSystem.get(spark.sparkContext.hadoopConfiguration)
    val path =  new Path(dataSourcePath)
    if(!fs.exists(path) && !fs.isDirectory(path)) {
        fs.mkdirs(path)
    }
}

createDirectory(transformDeltaOutputPath)
createDirectory(avgDeltaOutputKPIPath)
createDirectory(avgMoMKPIChangePath)

Criar dados de formato delta a partir do formato Parquet

Os dados de entrada são provenientes de listOfDataFile, de onde os dados foram descarregados de https://www.nyc.gov/site/tlc/about/tlc-trip-record-data.page.
Para demonstrar a viagem no tempo e a versão, carregue os dados individualmente
Execute a transformação e calcule o seguinte KPI de negócios em carga incremental:
1. A distância média
2. O custo médio por milha
3. O custo médio

Salve dados transformados e KPI no formato delta

import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.DataFrame

// UDF to compute sum of value paid by customer
def totalCustPaid = udf((basePassengerFare:Double, tolls:Double,bcf:Double,salesTax:Double,congSurcharge:Double,airportFee:Double, tips:Double) => {
    val total = basePassengerFare + tolls + bcf + salesTax + congSurcharge + airportFee + tips
    total
})

// read parquet file from spark conf with given file input
// transform data to compute total amount
// compute kpi for the given file/batch data
def readTransformWriteDelta(fileName:String, oldData:Option[DataFrame], format:String="parquet"):DataFrame = {
    val df = spark.read.format(format).load(fileName)
    val dfNewLoad= df.withColumn("total_amount",totalCustPaid($"base_passenger_fare",$"tolls",$"bcf",$"sales_tax",$"congestion_surcharge",$"airport_fee",$"tips"))
    // union with old data to compute KPI
    val dfFullLoad= oldData match {
        case Some(odf)=>
                dfNewLoad.union(odf)
        case _ =>
                dfNewLoad
    }
    dfFullLoad.createOrReplaceTempView("tempFullLoadCompute")
    val dfKpiCompute = spark.sql("SELECT round(avg(trip_miles),2) AS avgDist,round(avg(total_amount/trip_miles),2) AS avgCostPerMile,round(avg(total_amount),2) avgCost FROM tempFullLoadCompute")
    // save only new transformed data
    dfNewLoad.write.mode("overwrite").format("delta").save(transformDeltaOutputPath)
    //save compute KPI
    dfKpiCompute.write.mode("overwrite").format("delta").save(avgDeltaOutputKPIPath)
    // return incremental dataframe for next set of load
    dfFullLoad
}

// load data for each data file, use last dataframe for KPI compute with the current load
def loadData(dataFile: List[DataFile], oldDF:Option[DataFrame]):Boolean = {
    if(dataFile.isEmpty) {    
        true
    } else {
        val nextDataFile = dataFile.head
        val newFullDF = readTransformWriteDelta(nextDataFile.hdfsPath,oldDF)
        loadData(dataFile.tail,Some(newFullDF))
    }
}
val starTime=System.currentTimeMillis()
loadData(listOfDataFile,None)
println(s"Time taken in Seconds: ${(System.currentTimeMillis()-starTime)/1000}")

Ler o formato Delta usando a Tabela Delta

ler dados transformados
ler dados de KPI

import io.delta.tables._
val dtTransformed: io.delta.tables.DeltaTable = DeltaTable.forPath(transformDeltaOutputPath)
val dtAvgKpi: io.delta.tables.DeltaTable = DeltaTable.forPath(avgDeltaOutputKPIPath)

Esquema de impressão

Imprima o esquema da tabela delta para dados de KPI transformados e médios1.

// transform data schema
dtTransformed.toDF.printSchema
// Average KPI Data Schema
dtAvgKpi.toDF.printSchema

Captura de tela mostrando a saída do esquema de impressão.

Exibir o último KPI calculado da tabela de dados

dtAvgKpi.toDF.show(false)

Exibir histórico de KPI computado

Esta etapa exibe o histórico da tabela de transações KPI de _delta_log

dtAvgKpi.history().show(false)

Exibir dados de KPI após cada carregamento de dados

Usando a viagem no tempo, você pode visualizar as alterações de KPI após cada carregamento
Você pode armazenar todas as alterações de versão no formato CSV em avgMoMKPIChangePath , para que o Power BI possa ler essas alterações

val dfTxLog = spark.read.json(s"${transformDeltaOutputPath}/_delta_log/*.json")
dfTxLog.select(col("add")("path").alias("file_path")).withColumn("version",substring(input_file_name(),-6,1)).filter("file_path is not NULL").show(false)

Referência

Apache, Apache Spark, Spark e nomes de projetos de código aberto associados são marcas comerciais da Apache Software Foundation (ASF).

Partilhar via