Dela via

Kodexempel för Databricks Anslut för Python

  • Artikel

Kommentar

Den här artikeln beskriver Databricks Anslut för Databricks Runtime 13.0 och senare.

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

Kommentar

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

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

Du kan också använda följande enklare kodexempel för att experimentera med Databricks Anslut. Dessa exempel förutsätter att du använder standardautentisering för Databricks Anslut 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 Anslut 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)