Nota
O acesso a esta página requer autorização. Pode tentar iniciar sessão ou alterar os diretórios.
O acesso a esta página requer autorização. Pode tentar alterar os diretórios.
Este tutorial mostra-lhe como ligar o seu cluster Azure Databricks ao Azure Data Lake Storage e executar consultas e análises Spark sobre os seus dados. Também aprende a ingerir dados de voo não estruturados e transformá-los para o formato Apache Parquet para uma consulta eficiente.
Neste tutorial, você:
- Ingerir dados não estruturados numa conta de armazenamento
- Realize análises nos dados no Blob Storage
Se não tiver uma subscrição do Azure, crie uma conta gratuita antes de começar.
Pré-requisitos
Criar uma conta de armazenamento que tenha um namespace hierárquico (Armazenamento do Azure Data Lake)
Consulte Criar uma conta de armazenamento para usar com o Azure Data Lake Storage.
Verifique se sua conta de usuário tem a função de Colaborador de Dados de Blob de Armazenamento atribuída a ela.
Instale o AzCopy v10. Veja Transferir dados com AzCopy v10.
Crie um principal de serviço, crie um segredo de cliente e depois conceda ao principal de serviço acesso à conta de armazenamento.
Consulte Tutorial: Conectar-se ao Armazenamento do Azure Data Lake (Etapas 1 a 3). Depois de concluir essas etapas, certifique-se de colar a ID do locatário, a ID do aplicativo e os valores secretos do cliente em um arquivo de texto. Vai utilizá-los mais tarde no tutorial.
Criar um espaço de trabalho e um bloco de anotações do Azure Databricks
Crie um espaço de trabalho do Azure Databricks. Consulte Criar um espaço de trabalho do Azure Databricks.
Crie um bloco de notas. Consulte Criar um bloco de notas. Escolha Python como o idioma padrão do bloco de anotações.
Mantenha o seu bloco de notas aberto. Você utiliza-o nas seguintes secções.
Descarregar os dados de voo
Este tutorial utiliza dados de pontualidade de voo para janeiro de 2016 do Departamento de Estatísticas de Transportes dos EUA para demonstrar como executar uma operação ETL. Você deve baixar esses dados para concluir o tutorial.
Transfira o ficheiro On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip. Este ficheiro contém os dados do voo.
Descompacte o conteúdo do arquivo compactado e anote o nome do arquivo e o caminho do arquivo. Você precisa dessas informações em uma etapa posterior.
Para saber mais sobre a informação recolhida nos dados de desempenho de reporte atempado, consulte as descrições de campo no site do Bureau of Transportation Statistics.
Ingerir dados
Nesta secção, carregas os dados de voo .csv para a tua conta de Armazenamento do Azure Data Lake, e depois montas a conta de armazenamento no teu cluster Databricks. Finalmente, você usa o Databricks para ler os dados de voo no formato .csv e gravá-los de volta no armazenamento no formato Apache Parquet.
Carregue os dados do voo na sua conta de armazenamento
Use o AzCopy para copiar seu arquivo .csv para sua conta do Armazenamento do Azure Data Lake. Use o azcopy make comando para criar um contentor na sua conta de armazenamento. Depois usa o azcopy copy comando para copiar os dados csv que acabaste de descarregar para um diretório nesse contentor.
Nos passos seguintes, introduza os nomes do contentor que pretende criar, bem como do diretório e do blob para onde quer carregar os dados de voo no contentor. Use os nomes sugeridos em cada passo, ou especifique os seus próprios nomes seguindo as convenções de nomes para contentores, diretórios e blobs.
Abra uma janela de prompt de comandos e introduza o seguinte comando para iniciar sessão no Microsoft Entra ID e aceder à sua conta de armazenamento.
azcopy loginSiga as instruções que aparecem na janela do prompt de comando para autenticar sua conta de usuário.
Para criar um contêiner em sua conta de armazenamento para armazenar os dados de voo, digite o seguinte comando:
azcopy make "https://<storage-account-name>.dfs.core.windows.net/<container-name>"Substitua o valor do espaço reservado
<storage-account-name>pelo nome da sua conta de armazenamento.Substitua o
<container-name>marcador de posição por um nome para o contentor que pretende criar para armazenar os dados csv , como flight-data-container.
Para carregar (copiar) os dados csv para sua conta de armazenamento, digite o seguinte comando.
azcopy copy "<csv-folder-path>" https://<storage-account-name>.dfs.core.windows.net/<container-name>/<directory-name>/On_Time.csvSubstitua o valor de espaço reservado
<csv-folder-path>pelo caminho para o arquivo .csv .Substitua o valor do espaço reservado
<storage-account-name>pelo nome da sua conta de armazenamento.Substitua o espaço reservado
<container-name>pelo nome do contêiner em sua conta de armazenamento.Substitua o
<directory-name>marcador pelo nome de um diretório para armazenar os seus dados no contentor, como jan2016.
Monte sua conta de armazenamento no cluster Databricks
Nesta seção, monta o armazenamento de objetos na nuvem do Azure Data Lake Storage no Sistema de Arquivos Databricks (DBFS). Utilizas o princípio do serviço Microsoft Entra que criaste anteriormente para autenticação com a conta de armazenamento. Para obter mais informações, consulte Montagem do armazenamento de objetos na nuvem no Azure Databricks.
Anexe o bloco de notas ao cluster.
No bloco de notas que criou anteriormente, selecione o botão Ligar no canto superior direito da barra de ferramentas do bloco de notas. Este botão abre o seletor de recursos de computação. (Se já ligou o seu caderno a um cluster, o nome desse cluster aparece no texto do botão em vez de Conectar).
No menu suspenso do cluster, selecione qualquer cluster que tenha criado anteriormente.
O texto no seletor de cluster muda para começar. Aguarde até que o cluster termine de iniciar e que o nome do cluster apareça no botão antes de continuar.
Copie e cole o seguinte bloco de código na primeira célula, mas ainda não execute esse código.
configs = {"fs.azure.account.auth.type": "OAuth", "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider", "fs.azure.account.oauth2.client.id": "<appId>", "fs.azure.account.oauth2.client.secret": "<clientSecret>", "fs.azure.account.oauth2.client.endpoint": "https://login.microsoftonline.com/<tenantId>/oauth2/token", "fs.azure.createRemoteFileSystemDuringInitialization": "true"} dbutils.fs.mount( source = "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<directory-name>", mount_point = "/mnt/flightdata", extra_configs = configs)Neste bloco de código:
No
configs, substitua os valores de espaço reservado<appId>,<clientSecret>e<tenantId>pela ID do aplicativo, segredo do cliente e ID do locatário que você copiou ao criar o principal de serviço segundo os pré-requisitos.No URI
source, substitua os valores de espaço reservado<storage-account-name>,<container-name>e<directory-name>pelo nome da sua conta de armazenamento do Azure Data Lake, bem como pelo nome do contêiner e da diretoria que especificou quando carregou os dados de voo para a conta de armazenamento.Nota
O identificador de esquema no URI,
abfss, diz ao Databricks para usar o driver do Sistema de Ficheiros Blob do Azure com TLS (Transport Layer Security). Para saber mais sobre o URI, consulte Usar o URI de Armazenamento do Azure Data Lake.
Certifique-se de que o seu cluster conclui o arranque antes de prosseguir.
Pressione as teclas SHIFT + ENTER para executar o código neste bloco.
O contêiner e o diretório onde você carregou os dados de voo em sua conta de armazenamento agora estão acessíveis em seu notebook através do ponto de montagem, /mnt/flightdata.
Utilize o notebook do Databricks para converter CSV em Parquet
Agora que pode aceder aos dados de voos em formato csv através de um ponto de montagem do DBFS, utilize um DataFrame do Apache Spark para os carregar no seu espaço de trabalho e escreva-os novamente em formato Apache Parquet no armazenamento de objetos do Azure Data Lake Storage.
Um Spark DataFrame é uma estrutura de dados rotulada bidimensional com colunas de tipos potencialmente diferentes. Use um DataFrame para ler e escrever dados facilmente em vários formatos suportados. Com um DataFrame, você pode carregar dados do armazenamento de objetos na nuvem e executar análises e transformações dentro do cluster de computação sem afetar os dados subjacentes no armazenamento de objetos na nuvem. Para saber mais, consulte Trabalhar com DataFrames PySpark no Azure Databricks.
Apache Parquet é um formato de ficheiro colunar com otimizações que aceleram as consultas. É um formato de arquivo mais eficiente do que CSV ou JSON. Para saber mais, consulte Arquivos do Parquet.
No bloco de anotações, adicione uma nova célula e cole o código a seguir nela.
# Use the previously established DBFS mount point to read the data.
# Create a DataFrame to read the csv data.
# The header option specifies that the first row of data should be used as the DataFrame column names
# The inferschema option specifies that the column data types should be inferred from the data in the file
flight_df = spark.read.format('csv').options(
header='true', inferschema='true').load("/mnt/flightdata/*.csv")
# Read the airline csv file and write the output to parquet format for easy query.
flight_df.write.mode("append").parquet("/mnt/flightdata/parquet/flights")
print("Done")
Pressione as teclas SHIFT + ENTER para executar o código neste bloco.
Antes de avançar para a secção seguinte, certifique-se de que todos os dados do Parquet estão escritos e que "Feito" aparece na saída.
Explorar dados
Nesta secção, utilize a utilidade do sistema de ficheiros Databricks para explorar o seu Azure Data Lake Storage object storage, utilizando o ponto de montagem DBFS que criou na secção anterior.
Em uma nova célula, cole o código a seguir para obter uma lista dos arquivos no ponto de montagem. O primeiro comando gera uma lista de arquivos e diretórios. O segundo comando exibe a saída em formato tabular para facilitar a leitura.
dbutils.fs.ls("/mnt/flightdata")
display(dbutils.fs.ls("/mnt/flightdata"))
Pressione as teclas SHIFT + ENTER para executar o código neste bloco.
Observe que o diretório parquet aparece na listagem. Você salvou os dados de voo em formato .csv no formato parquet no diretório parquet/flights na seção anterior. Para listar arquivos no diretório parquet/flights , cole o seguinte código em uma nova célula e execute-o:
display(dbutils.fs.ls("/mnt/flightdata/parquet/flights"))
Para criar um novo arquivo e listá-lo, cole o seguinte código em uma nova célula e execute-o:
dbutils.fs.put("/mnt/flightdata/mydirectory/mysubdirectory/1.txt", "Hello, World!", True)
display(dbutils.fs.ls("/mnt/flightdata/mydirectory/mysubdirectory"))
Como você não precisa do arquivo 1.txt neste tutorial, você pode colar o código a seguir em uma célula e executá-lo para excluir recursivamente mydirectory. O True parâmetro indica uma exclusão recursiva.
dbutils.fs.rm("/mnt/flightdata/mydirectory", True)
Por conveniência, use o comando ajuda para aprender detalhes sobre outros comandos.
dbutils.fs.help("rm")
Ao usar estes exemplos de código, explorou a natureza hierárquica do HDFS ao utilizar dados armazenados numa conta de armazenamento com o Azure Data Lake Storage ativado.
Consultar os dados
De seguida, consulta os dados que carregaste na tua conta de armazenamento. Insira cada um dos seguintes blocos de código em uma nova célula e pressione SHIFT + ENTER para executar o script Python.
DataFrames fornecem um rico conjunto de funções (selecionar colunas, filtrar, juntar, agregar) que permitem resolver problemas comuns de análise de dados de forma eficiente.
Para carregar um DataFrame a partir dos seus dados de voo guardados previamente no formato parquet e explorar algumas das funcionalidades suportadas, insira este script numa nova célula e execute-o.
# Read the existing parquet file for the flights database that was created earlier
flight_df = spark.read.parquet("/mnt/flightdata/parquet/flights")
# Print the schema of the dataframe
flight_df.printSchema()
# Print the flight database size
print("Number of flights in the database: ", flight_df.count())
# Show the first 25 rows (20 is the default)
# To show the first n rows, run: df.show(n)
# The second parameter indicates that column lengths shouldn't be truncated (default is 20 characters)
flight_df.show(25, False)
# You can also use the DataFrame to run simple queries. Results are returned in a DataFrame.
# Show the first 25 rows of the results of a query that returns selected columns for all flights originating from airports in Texas
flight_df.select("FlightDate", "Reporting_Airline", "Flight_Number_Reporting_Airline", "OriginCityName", "DepTime", "DestCityName", "ArrTime", "ArrDelay").filter("OriginState = 'TX'").show(258, False)
# Use display to run visualizations
# Preferably run this in a separate cmd cell
display(flight_df)
Insira esse script em uma nova célula para executar algumas consultas básicas de análise nos dados. Você pode optar por executar o script inteiro (SHIFT + ENTER), realçar cada consulta e executá-la separadamente com CTRL + SHIFT + ENTER, ou inserir cada consulta em uma célula separada e executá-la lá.
# create a temporary sql view for querying flight information
flight_data = spark.read.parquet('/mnt/flightdata/parquet/flights')
flight_data.createOrReplaceTempView('FlightTable')
# Print the total number of flights in Jan 2016 (the number of rows in the flight data).
print("Number of flights in Jan 2016: ", flight_data.count())
# Using spark sql, query the parquet file to return the total flights of each airline
num_flights_by_airline=spark.sql("SELECT Reporting_Airline, count(*) AS NumFlights FROM FlightTable GROUP BY Reporting_Airline ORDER BY NumFlights DESC")
num_flights_by_airline.show()
# List out all the airports in Texas
airports_in_texas = spark.sql(
"SELECT DISTINCT(OriginCityName) FROM FlightTable WHERE OriginStateName = 'Texas'")
print('Airports in Texas: ', airports_in_texas.count())
airports_in_texas.show(100, False)
# Find all airlines that fly from Texas
airlines_flying_from_texas = spark.sql(
"SELECT DISTINCT(Reporting_Airline) FROM FlightTable WHERE OriginStateName='Texas'")
print('Airlines that fly to/from Texas: ', airlines_flying_from_texas.count())
airlines_flying_from_texas.show(100, False)
# List airlines by average arrival delay (negative values indicate early flights)
avg_arrival_delay=spark.sql(
"SELECT Reporting_Airline, count(*) AS NumFlights, avg(DepDelay) AS AverageDepDelay, avg(ArrDelay) AS AverageArrDelay FROM FlightTable GROUP BY Reporting_Airline ORDER BY AverageArrDelay DESC")
print("Airlines by average arrival delay")
avg_arrival_delay.show()
# List airlines by the highest percentage of delayed flights. A delayed flight is one with a departure or arrival delay that is greater than 15 minutes
spark.sql("DROP VIEW IF EXISTS totalFlights")
spark.sql("DROP VIEW IF EXISTS delayedFlights")
spark.sql(
"CREATE TEMPORARY VIEW totalFlights AS SELECT Reporting_Airline, count(*) AS NumFlights FROM FlightTable GROUP BY Reporting_Airline")
spark.sql(
"CREATE TEMPORARY VIEW delayedFlights AS SELECT Reporting_Airline, count(*) AS NumDelayedFlights FROM FlightTable WHERE DepDelay>15 or ArrDelay>15 GROUP BY Reporting_Airline")
percent_delayed_flights=spark.sql(
"SELECT totalFlights.Reporting_Airline, totalFlights.NumFlights, delayedFlights.NumDelayedFlights, delayedFlights.NumDelayedFlights/totalFlights.NumFlights*100 AS PercentFlightsDelayed FROM totalFlights INNER JOIN delayedFlights ON totalFlights.Reporting_Airline = delayedFlights.Reporting_Airline ORDER BY PercentFlightsDelayed DESC")
print("Airlines by percentage of flights delayed")
percent_delayed_flights.show()
Resumo
Neste tutorial, você:
Criei recursos Azure, incluindo uma conta de armazenamento Azure Data Lake Storage e um principal de serviço Microsoft Entra, e atribuí permissões para aceder à conta de armazenamento.
Criou um espaço de trabalho e um bloco de anotações do Azure Databricks.
AzCopy foi usado para carregar dados de voo não estruturados no formato .csv para a conta de armazenamento do Azure Data Lake.
Utilizou funções do utilitário do Sistema de Arquivos Databricks para montar a sua conta de armazenamento do Azure Data Lake Storage e explorar o seu sistema de arquivos hierárquico.
Utilizou os Apache Spark DataFrames para transformar os seus dados de voo em .csv para o formato Apache Parquet e armazená-los de volta na sua conta de armazenamento do Azure Data Lake Storage.
DataFrames usados para explorar os dados de voo e realizar uma consulta simples.
Usado o Apache Spark SQL para consultar os dados de voo para o número total de voos de cada companhia aérea em janeiro de 2016, os aeroportos no Texas, as companhias aéreas que voam do Texas, o atraso médio de chegada em minutos para cada companhia aérea nacionalmente e a porcentagem de voos de cada companhia aérea que atrasaram partidas ou chegadas.
Limpar recursos
Para preservar o notebook e evitar encargos, desligue (termine) o cluster. Selecione-o no seletor de computação localizado no canto superior direito da barra de ferramentas do caderno, selecione Terminar no menu e confirme a sua seleção. (Por defeito, o cluster termina automaticamente após 120 minutos de inatividade.)
Para eliminar recursos individuais do espaço de trabalho, como cadernos e clusters, use a barra lateral esquerda do espaço de trabalho. Para obter instruções detalhadas, consulte Excluir um cluster ou Excluir um bloco de anotações.
Quando não forem mais necessários, exclua o grupo de recursos e todos os recursos relacionados. Para fazer isso no portal do Azure, selecione o grupo de recursos para a conta de armazenamento e o espaço de trabalho e selecione Excluir.