Kodexempel för Databricks Connect för Python

Kommentar

Den här artikeln beskriver Databricks Connect för Databricks Runtime 13.3 LTS och senare.

Den här artikeln innehåller kodexempel som använder Databricks Connect för Python. Med Databricks Connect kan du ansluta populära IDE:er, notebook-servrar och anpassade program till Azure Databricks-kluster. Se Vad är Databricks Connect?. För Scala-versionen av den här artikeln, se Kodexempel för Databricks Connect för Scala.

Kommentar

Innan du börjar använda Databricks Connect måste du konfigurera Databricks Connect-klienten.

Databricks innehåller flera ytterligare exempelprogram som visar hur du använder Databricks Connect. Se exempelprogram för Databricks Connect-lagringsplatsen i GitHub, specifikt:

• Ett enkelt ETL-program
• Ett interaktivt dataprogram baserat på Plotly
• Ett interaktivt dataprogram baserat på Plotly och PySpark AI

Du kan också använda följande enklare kodexempel för att experimentera med Databricks Connect. Dessa exempel förutsätter att du använder standardautentisering för Databricks Connect-klientkonfiguration.

Det här enkla kodexemplet frågar den angivna tabellen och visar sedan den angivna tabellens första 5 rader. Om du vill använda en annan tabell justerar du anropet till spark.read.table.

from databricks.connect import DatabricksSession

spark = DatabricksSession.builder.getOrCreate()

df = spark.read.table("samples.nyctaxi.trips")
df.show(5)

Det här längre kodexemplet gör följande:

1. Skapar en minnesintern DataFrame.
2. Skapar en tabell med namnet zzz_demo_temps_table i default schemat. Om tabellen med det här namnet redan finns tas tabellen bort först. Om du vill använda ett annat schema eller en annan tabell justerar du anropen till spark.sql, temps.write.saveAsTableeller båda.
3. Sparar dataramens innehåll i tabellen.
4. Kör en SELECT fråga i tabellens innehåll.
5. Visar frågans resultat.
6. Tar bort tabellen.

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')

Kommentar

I följande exempel beskrivs hur du skriver kod som är portabel mellan Databricks Connect för Databricks Runtime 13.3 LTS och senare i miljöer där DatabricksSession klassen inte är tillgänglig.

I följande exempel används DatabricksSession klassen, eller klassen SparkSession om DatabricksSession klassen inte är tillgänglig, för att köra frågor mot den angivna tabellen och returnera de första 5 raderna. I det SPARK_REMOTE här exemplet används miljövariabeln för autentisering.

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)