مشاركة عبر

أمثلة التعليمات البرمجية ل Databricks Connect ل Python

  • مقالة

إشعار

تتناول هذه المقالة Databricks Connect ل Databricks Runtime 13.3 LTS وما فوق.

توفر هذه المقالة أمثلة التعليمات البرمجية التي تستخدم Databricks Connect ل Python. يمكنك Databricks Connect من توصيل IDEs الشائعة وخوادم دفاتر الملاحظات والتطبيقات المخصصة إلى مجموعات Azure Databricks. راجع ما هو Databricks Connect؟. للحصول على إصدار Scala من هذه المقالة، راجع أمثلة التعليمات البرمجية ل Databricks Connect ل Scala.

إشعار

قبل البدء في استخدام Databricks Connect، يجب عليك إعداد عميل Databricks Connect.

يوفر Databricks العديد من التطبيقات الإضافية التي توضح كيفية استخدام Databricks Connect. راجع أمثلة التطبيقات لمستودع Databricks Connect في GitHub، على وجه التحديد:

يمكنك أيضا استخدام أمثلة التعليمات البرمجية الأبسط التالية لتجربة Databricks Connect. تفترض هذه الأمثلة أنك تستخدم المصادقة الافتراضية لإعداد عميل Databricks Connect.

يستعلم مثال التعليمات البرمجية البسيط هذا عن الجدول المحدد ثم يعرض أول 5 صفوف للجدول المحدد. لاستخدام جدول مختلف، اضبط الاستدعاء إلى spark.read.table.

from databricks.connect import DatabricksSession

spark = DatabricksSession.builder.getOrCreate()

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

يقوم مثال التعليمات البرمجية الأطول هذا بالآتي:

  1. إنشاء DataFrame في الذاكرة.
  2. إنشاء جدول بالاسم zzz_demo_temps_table داخل default المخطط. إذا كان الجدول بهذا الاسم موجودا بالفعل، يتم حذف الجدول أولا. لاستخدام مخطط أو جدول مختلف، اضبط الاستدعاءات إلى spark.sqlأو temps.write.saveAsTableأو كليهما.
  3. يحفظ محتويات DataFrame في الجدول.
  4. SELECT تشغيل استعلام على محتويات الجدول.
  5. إظهار نتيجة الاستعلام.
  6. حذف الجدول.
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')

إشعار

يصف المثال التالي كيفية كتابة التعليمات البرمجية المحمولة بين Databricks Connect ل Databricks Runtime 13.3 LTS والإصدارات الأحدث في البيئات التي لا تتوفر فيها DatabricksSession الفئة.

يستخدم DatabricksSession المثال التالي الفئة ، أو يستخدم SparkSession الفئة إذا كانت DatabricksSession الفئة غير متوفرة، للاستعلام عن الجدول المحدد وإرجاع أول 5 صفوف. يستخدم SPARK_REMOTE هذا المثال متغير البيئة للمصادقة.

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)