Scala için Databricks Connect için kod örnekleri

Not

Bu makale Databricks Runtime 13.3 LTS ve üzeri için Databricks Connect'i kapsar.

Bu makalede Scala için Databricks Connect kullanan kod örnekleri verilmektedir. Databricks Connect popüler IDE'leri, not defteri sunucularını ve özel uygulamaları Azure Databricks kümelere bağlamanızı sağlar. Bkz. Databricks Connect. Bu makalenin Python versiyonu için bkz. Python için Databricks Connect Kod örnekleri.

Databricks Connect'i kullanmaya başlamadan önce Databricks Connect istemcisini ayarlamanız gerekir.

Aşağıdaki örneklerde Databricks Connect istemci kurulumu için varsayılan kimlik doğrulamasını kullandığınız varsayılır.

Örnek: Tablo okuma

Bu basit kod örneği, belirtilen tabloyu sorgular ve ardından belirtilen tablonun ilk 5 satırını gösterir.

import com.databricks.connect.DatabricksSession
import org.apache.spark.sql.SparkSession

object Main {
  def main(args: Array[String]): Unit = {
    val spark = DatabricksSession.builder().getOrCreate()
    val df = spark.read.table("samples.nyctaxi.trips")
    df.limit(5).show()
  }
}

DataFrame oluşturma

Aşağıdaki kod örneği:

  1. Bellek içi bir DataFrame oluşturur.
  2. Şema içinde zzz_demo_temps_table adıyla default bir tablo oluşturur. Bu isimle bir tablo zaten mevcutsa, önce bu tablo silinir. Farklı bir şema veya tablo kullanmak için çağrıları spark.sql, temps.write.saveAsTable veya her ikisine ayarlayın.
  3. DataFrame'in içeriğini tabloya kaydeder.
  4. Tablonun içeriğinde bir SELECT sorgu çalıştırır.
  5. Sorgunun sonucunu gösterir.
  6. Tabloyu siler.
import com.databricks.connect.DatabricksSession
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
import java.time.LocalDate

object Main {
  def main(args: Array[String]): Unit = {
    val spark = DatabricksSession.builder().getOrCreate()

    // Create a Spark DataFrame consisting of high and low temperatures
    // by airport code and date.
    val schema = StructType(
      Seq(
        StructField("AirportCode", StringType, false),
        StructField("Date", DateType, false),
        StructField("TempHighF", IntegerType, false),
        StructField("TempLowF", IntegerType, false)
      )
    )

    val data = Seq(
      ( "BLI", LocalDate.of(2021, 4, 3), 52, 43 ),
      ( "BLI", LocalDate.of(2021, 4, 2), 50, 38),
      ( "BLI", LocalDate.of(2021, 4, 1), 52, 41),
      ( "PDX", LocalDate.of(2021, 4, 3), 64, 45),
      ( "PDX", LocalDate.of(2021, 4, 2), 61, 41),
      ( "PDX", LocalDate.of(2021, 4, 1), 66, 39),
      ( "SEA", LocalDate.of(2021, 4, 3), 57, 43),
      ( "SEA", LocalDate.of(2021, 4, 2), 54, 39),
      ( "SEA", LocalDate.of(2021, 4, 1), 56, 41)
    )

    val temps = spark.createDataFrame(data).toDF(schema.fieldNames: _*)

    // 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.
    val 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")
  }
}

Örnek: DatabricksSesssion veya SparkSession kullanma

Aşağıdaki örnekte, Databricks Connect'teki sınıfın SparkSession kullanılamadığı durumlarda sınıfın DatabricksSession nasıl kullanılacağı açıklanmaktadır.

Bu örnek, belirtilen tabloyu sorgular ve ilk 5 satırı döndürür. Bu örnekte kimlik doğrulaması için ortam değişkeni kullanılır SPARK_REMOTE .

import org.apache.spark.sql.{DataFrame, SparkSession}

object Main {
  def main(args: Array[String]): Unit = {
    getTaxis(getSpark()).show(5)
  }

  private def getSpark(): SparkSession = {
    SparkSession.builder().getOrCreate()
  }

  private def getTaxis(spark: SparkSession): DataFrame = {
    spark.read.table("samples.nyctaxi.trips")
  }
}

Ek kaynaklar

Databricks, Databricks Connect GitHub deposunda Databricks Connect'in nasıl kullanılacağını gösteren ek örnek uygulamalar sağlar, örneğin:

  • Last updated on