Observação
O acesso a essa página exige autorização. Você pode tentar entrar ou alterar diretórios.
O acesso a essa página exige autorização. Você pode tentar alterar os diretórios.
Este tutorial mostra como conectar seu cluster Azure Databricks a Azure Data Lake Storage e executar consultas e análises do Spark em seus dados. Você também aprenderá a ingerir dados de voo não estruturados e transformá-los no formato Apache Parquet para consulta eficiente.
Neste tutorial, você:
- Ingerir dados não estruturados em uma conta de armazenamento
- Executar a análise em seus dados no Armazenamento de Blobs
Se você não tiver uma assinatura do Azure, crie uma conta gratuita antes de começar.
Pré-requisitos
Criar uma conta de armazenamento que tenha um namespace hierárquico (Azure Data Lake Storage)
Confira Criar uma conta de armazenamento para uso com o Azure Data Lake Storage.
Verifique se a sua conta de usuário tem a função Colaborador de Dados do Storage Blob atribuída a ela.
Instale o AzCopy v10. Consulte Transferir dados com AzCopy v10.
Crie uma entidade de serviço, crie um segredo do cliente e, em seguida, conceda à entidade de serviço acesso à conta de armazenamento.
Confira Tutorial: Conectar-se ao Azure Data Lake Storage (etapas 1 a 3). Depois de concluir essas etapas, cole a ID do locatário, a ID do aplicativo e os valores do segredo do cliente em um arquivo de texto. Você os usará posteriormente neste tutorial.
Crie um workspace e um notebook no Azure Databricks
Criar um workspace do Azure Databricks. Consulte Criar um espaço de trabalho do Azure Databricks.
Crie um notebook. Consulte Criar um notebook. Escolha Python como o idioma padrão do notebook.
Mantenha seu notebook aberto. Use-o nas seções a seguir.
Baixar os dados de voos
Este tutorial usa os dados de desempenho de pontualidade de voo de janeiro de 2016 da Agência de Estatísticas de Transporte, para demonstrar como executar uma operação de ETL. Você precisa baixar esses dados para concluir o tutorial.
Baixe o arquivo On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip. Este arquivo contém os dados de voo.
Descompacte o conteúdo do arquivo compactado e anote o nome e o caminho do arquivo. Você precisará dessas informações em uma etapa posterior.
Para saber mais sobre as informações capturadas nos dados de desempenho de relatórios pontuais, consulte as descrições de campo no site do Bureau of Transportation Statistics.
Ingerir dados
Nesta seção, você carrega os dados de voo .csv para a sua conta do Azure Data Lake Storage e, em seguida, monta a conta de armazenamento no cluster do Databricks. Por fim, usará o Databricks para ler os dados de voo .csv e gravá-los de volta ao armazenamento no formato Apache parquet.
Carregue os dados de voo na conta de armazenamento
Use o AzCopy para copiar o arquivo .csv para a sua conta do Azure Data Lake Storage. Use o azcopy make comando para criar um contêiner em sua conta de armazenamento. Em seguida, use o azcopy copy comando para copiar os dados CSV que você acabou de baixar para um diretório nesse contêiner.
Nas etapas a seguir, insira nomes para o contêiner que você deseja criar e o diretório e o blob para o qual você deseja carregar os dados de voo no contêiner. Use os nomes sugeridos em cada etapa ou especifique seus próprios nomes ao seguir as convenções de nomenclatura para contêineres, diretórios e blobs.
Abra uma janela do prompt de comando e insira o comando a seguir para entrar no Microsoft Entra ID para acessar sua conta de armazenamento.
azcopy loginSiga as instruções exibidas na janela do prompt de comando para autenticar sua conta de usuário.
Para criar um contêiner na conta de armazenamento a fim de armazenar os dados de voo, insira 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 de sua conta de armazenamento.Substitua o marcador
<container-name>por um nome para o contêiner que você deseja criar para armazenar os dados em csv, como flight-data-container.
Para carregar (copiar) os dados .csv em sua conta de armazenamento, insira o comando a seguir.
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 nome caminho no arquivo .csv.Substitua o valor do espaço reservado
<storage-account-name>pelo nome de sua conta de armazenamento.Substitua o espaço reservado
<container-name>pelo nome do contêiner em sua conta de armazenamento.Substitua o marcador
<directory-name>pelo nome de um diretório para armazenar dados no contêiner, como jan2016.
Montar sua conta de armazenamento no cluster do Databricks
Nesta seção, você montará o armazenamento de objetos de nuvem do Azure Data Lake Storage no Databricks File System (DBFS). Use a entidade de serviço Microsoft Entra criada anteriormente para autenticação com a conta de armazenamento. Para obter mais informações, confira Montagem do armazenamento de objetos de nuvem no Azure Databricks.
Anexe seu notebook ao cluster.
No notebook criado anteriormente, selecione o botão Conectar no canto superior direito da barra de ferramentas do notebook. Este botão abre o seletor de recursos de computação. (Se você já tiver conectado seu notebook a um cluster, o nome desse cluster será exibido no texto do botão em vez de Conectar).
No menu suspenso do cluster, selecione qualquer cluster criado anteriormente.
O texto no seletor de cluster é alterado para iniciar. Aguarde a conclusão da inicialização do cluster e o nome do cluster apareça no botão antes de continuar.
Copie e cole o bloco de código a seguir na primeira célula, mas não execute esse código ainda.
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:
Em
configs, substitua os valores de espaço reservado<appId>,<clientSecret>e<tenantId>, pela ID do aplicativo, pelo segredo do cliente e pela ID do locatário copiado quando criou a entidade de serviço nos pré-requisitos.Na URI
source, substitua os valores de marcador<storage-account-name>,<container-name>e<directory-name>pelos nomes da sua conta de armazenamento do Azure Data Lake Storage e do contêiner e diretório que você especificou quando carregou os dados de voo para a conta de armazenamento.Observação
O identificador de esquema no URI,
abfss, informa ao Databricks para usar o driver do Azure Blob File System com o TLS (protocolo TLS). Para saber mais sobre o URI, confira Usar o URI do Azure Data Lake Storage.
Certifique-se de que o cluster termine de iniciar antes de prosseguir.
Pressione as teclas SHIFT+ENTER para executar o código nesse bloco.
O contêiner e o diretório onde você carregou os dados de voo na sua conta de armazenamento agora estão acessíveis no seu notebook por meio do ponto de montagem /mnt/flightdata.
Usar o notebook do Databricks para converter CSV em Parquet
Agora que você pode acessar os dados de voo em csv por meio de um ponto de montagem do DBFS, use um DataFrame do Apache Spark para carregá-los em seu espaço de trabalho e gravá-los de volta no formato Apache Parquet no armazenamento de objetos do Azure Data Lake Storage.
Um DataFrame do Spark é uma estrutura de dados rotulada bidimensional com colunas de tipos potencialmente diferentes. Use um DataFrame para ler e gravar dados facilmente em vários formatos com suporte. Com um DataFrame, é possível carregar dados do armazenamento de objetos na nuvem e executar análises e transformações neles dentro do cluster de computação, sem afetar os dados subjacentes no armazenamento de objetos na nuvem. Para saber mais, confira Trabalhar com o DataFrames do PySpark no Azure Databricks.
O Apache Parquet é um formato de arquivo columnar com otimizações que aceleram as consultas. É um formato de arquivo mais eficiente do que CSV ou JSON. Para saber mais, confira Arquivos Parquet.
No notebook, adicione uma nova célula e cole o código a seguir nele.
# 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 nesse bloco.
Antes de prosseguir para a próxima seção, verifique se todos os dados Parquet tenham sido gravados e que "Concluído" apareça na saída.
Explorar dados
Nesta seção, use o utilitário do sistema de arquivos do Databricks para explorar seu armazenamento de objetos Azure Data Lake Storage usando o ponto de montagem do DBFS criado na seçã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 no 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 nesse bloco.
Observe que o diretório parquet aparece na listagem. Você salvou os dados de voo .csv no formato parquet no diretório parquet/flights na seção anterior. Para listar arquivos no diretório parquet/voos, 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 arquivo 1.txt não é necessário nesse tutorial, é possível colar o código a seguir em uma célula e executá-lo para excluir o mydirectory de forma recursiva. O parâmetro True indica uma exclusão recursiva.
dbutils.fs.rm("/mnt/flightdata/mydirectory", True)
Como conveniência, use o comando de ajuda para aprender detalhes sobre outros comandos.
dbutils.fs.help("rm")
Usando esses exemplos de código, você explorou a natureza hierárquica do HDFS usando dados armazenados em uma conta de armazenamento com Azure Data Lake Storage habilitados.
Consultar os dados
Em seguida, consulte os dados que você carregou em sua conta de armazenamento. Insira cada um dos blocos de código a seguir em uma nova célula e pressione SHIFT + ENTER para executar o script de Python.
Os DataFrames contém um amplo conjunto de funções (selecionar colunas, filtrar, unir, agregar) para resolução de problemas comuns de análise de dados com eficiência.
Para carregar um DataFrame a partir dos dados de voo do parquet salvos anteriormente e explorar algumas das funcionalidades com suporte, insira esse script em uma 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. É possível 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ê:
Criou recursos do Azure, incluindo uma conta de armazenamento do Azure Data Lake Storage e uma entidade de serviço do Microsoft Entra, e atribuiu permissões para acessar a conta de armazenamento.
Criou um espaço de trabalho e um notebook do Azure Databricks.
Usou o AzCopy para fazer upload de dados de voo não estruturados em .csv para a conta de armazenamento do Azure Data Lake Storage.
As funções do utilitário do Databricks File System foram usadas para montar a sua conta de armazenamento do Azure Data Lake Storage e explorar seu sistema de arquivos hierárquico.
Utilizou os DataFrames do Apache Spark para transformar seus dados de voo .csv em formato Apache Parquet e armazená-los novamente na sua conta de armazenamento do Azure Data Lake Storage.
Usou DataFrames para explorar os dados de voo e realizar uma consulta simples.
Usou o Apache Spark do 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 do Texas, as companhias aéreas que voam do Texas, o atraso médio à 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 os recursos
Para preservar o notebook e evitar encargos, desligue (encerre) seu cluster. Selecione-o no seletor de computação localizado no canto superior direito da barra de ferramentas do notebook, selecione Encerrar no menu e confirme sua seleção. (Por padrão, o cluster é encerrado automaticamente após 120 minutos de inatividade.)
Para excluir recursos individuais desse espaço de trabalho, como notebooks e clusters, use a barra lateral esquerda do espaço de trabalho. Para obter instruções detalhadas, consulte Excluir um cluster ou Excluir um notebook.
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 da conta de armazenamento e o workspace e selecione Excluir.