Exemplos de código para Databricks Connect for Python
Este artigo abrange o Databricks Connect para Databricks Runtime 13.3 LTS e acima.
Este artigo fornece exemplos de código que usam Databricks Connect para Python. Databricks Connect permite que o senhor conecte os populares IDEs, servidores de notebook e aplicativos personalizados ao clustering Databricks. Consulte O que é o Databricks Connect? Para obter a versão Scala deste artigo, consulte Exemplos de código para Databricks Connect para Scala.
Antes de começar a usar o Databricks Connect, o senhor deve configurar o cliente Databricks Connect.
A Databricks fornece vários aplicativos de exemplo adicionais que mostram como usar o Databricks Connect. Veja os aplicativos de exemplo para o repositório do Databricks Connect no GitHub, especificamente:
- Uma aplicação ETL simples
- Uma aplicação de dados interativa baseada em Plotly
- Uma aplicação de dados interativa baseado em Plotly e PySpark AI
O senhor também pode usar os seguintes exemplos de código mais simples para experimentar o Databricks Connect. Esses exemplos pressupõem que o senhor esteja usando a autenticação default para a configuração do cliente Databricks Connect.
Esse exemplo de código simples consulta a tabela especificada e, em seguida, mostra as primeiras 5 linhas da tabela especificada. Para usar uma tabela diferente, ajuste a chamada para spark.read.table
.
from databricks.connect import DatabricksSession
spark = DatabricksSession.builder.getOrCreate()
df = spark.read.table("samples.nyctaxi.trips")
df.show(5)
Esse exemplo de código mais longo faz o seguinte:
- Cria um DataFrame na memória.
- Cria uma tabela com o nome
zzz_demo_temps_table
no esquemadefault
. Se a tabela com esse nome já existir, ela será excluída primeiro. Para usar um esquema ou tabela diferente, ajuste as chamadas paraspark.sql
,temps.write.saveAsTable
ou ambos. - Salva o conteúdo do DataFrame na tabela.
- executar uma consulta
SELECT
sobre o conteúdo da tabela. - Mostra o resultado da consulta.
- Exclui a tabela.
from databricks.connect import DatabricksSession
from pyspark.sql.types import *
from datetime import date
spark = DatabricksSession.builder.getOrCreate()
# Create a Spark DataFrame consisting of high and low temperatures
# by airport code and date.
schema = StructType([
StructField('AirportCode', StringType(), False),
StructField('Date', DateType(), False),
StructField('TempHighF', IntegerType(), False),
StructField('TempLowF', IntegerType(), False)
])
data = [
[ 'BLI', date(2021, 4, 3), 52, 43],
[ 'BLI', date(2021, 4, 2), 50, 38],
[ 'BLI', date(2021, 4, 1), 52, 41],
[ 'PDX', date(2021, 4, 3), 64, 45],
[ 'PDX', date(2021, 4, 2), 61, 41],
[ 'PDX', date(2021, 4, 1), 66, 39],
[ 'SEA', date(2021, 4, 3), 57, 43],
[ 'SEA', date(2021, 4, 2), 54, 39],
[ 'SEA', date(2021, 4, 1), 56, 41]
]
temps = spark.createDataFrame(data, schema)
# Create a table on the Databricks cluster and then fill
# the table with the DataFrame's contents.
# If the table already exists from a previous run,
# delete it first.
spark.sql('USE default')
spark.sql('DROP TABLE IF EXISTS zzz_demo_temps_table')
temps.write.saveAsTable('zzz_demo_temps_table')
# Query the table on the Databricks cluster, returning rows
# where the airport code is not BLI and the date is later
# than 2021-04-01. Group the results and order by high
# temperature in descending order.
df_temps = spark.sql("SELECT * FROM zzz_demo_temps_table " \
"WHERE AirportCode != 'BLI' AND Date > '2021-04-01' " \
"GROUP BY AirportCode, Date, TempHighF, TempLowF " \
"ORDER BY TempHighF DESC")
df_temps.show()
# Results:
#
# +-----------+----------+---------+--------+
# |AirportCode| Date|TempHighF|TempLowF|
# +-----------+----------+---------+--------+
# | PDX|2021-04-03| 64| 45|
# | PDX|2021-04-02| 61| 41|
# | SEA|2021-04-03| 57| 43|
# | SEA|2021-04-02| 54| 39|
# +-----------+----------+---------+--------+
# Clean up by deleting the table from the Databricks cluster.
spark.sql('DROP TABLE zzz_demo_temps_table')
O exemplo a seguir descreve como escrever um código que seja portátil entre Databricks Connect e Databricks Runtime 13.3 LTS e acima em ambientes em que a classe DatabricksSession
não está disponível.
O exemplo a seguir usa a classe DatabricksSession
, ou usa a classe SparkSession
se a classe DatabricksSession
não estiver disponível, para consultar a tabela especificada e retornar as primeiras 5 linhas. Este exemplo usa a variável de ambiente SPARK_REMOTE
para autenticação.
from pyspark.sql import SparkSession, DataFrame
def get_spark() -> SparkSession:
try:
from databricks.connect import DatabricksSession
return DatabricksSession.builder.getOrCreate()
except ImportError:
return SparkSession.builder.getOrCreate()
def get_taxis(spark: SparkSession) -> DataFrame:
return spark.read.table("samples.nyctaxi.trips")
get_taxis(get_spark()).show(5)