Codevoorbeelden voor Databricks Connect voor Scala

Dit artikel bevat codevoorbeelden die Gebruikmaken van Databricks Connect voor Scala. Met Databricks Connect kunt u populaire IDE's, notebookservers en aangepaste toepassingen verbinden met Azure Databricks-clusters. Zie Wat is Databricks Connect? Zie Codevoorbeelden voor Databricks Connect voor Python voor de Python-versie van dit artikel.

Voordat u Databricks Connect gaat gebruiken, moet u de Databricks Connect-client instellen.

In de volgende voorbeelden wordt ervan uitgegaan dat u standaardverificatie gebruikt voor de installatie van de Databricks Connect-client.

Voorbeeld: Een tabel lezen

In dit eenvoudige codevoorbeeld wordt een query uitgevoerd op de opgegeven tabel en worden vervolgens de eerste vijf rijen van de opgegeven tabel weergegeven.

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()
  }
}

Een DataFrame maken

Dit codevoorbeeld hieronder:

1. Hiermee maakt u een DataFrame in het geheugen.
2. Hiermee maakt u een tabel met de naam zzz_demo_temps_table binnen het default schema. Als de tabel met deze naam al bestaat, wordt de tabel eerst verwijderd. Als u een ander schema of een andere tabel wilt gebruiken, past u de aanroepen aan naar spark.sql, temps.write.saveAsTableof beide.
3. Slaat de inhoud van het DataFrame op in de tabel.
4. Voert een SELECT query uit op de inhoud van de tabel.
5. Geeft het resultaat van de query weer.
6. Verwijdert de tabel.

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")
  }
}

Voorbeeld: DatabricksSesssion of SparkSession gebruiken

In het volgende voorbeeld wordt beschreven hoe u de SparkSession-klasse gebruikt in gevallen waarin de klasse DatabricksSession in Databricks Connect niet beschikbaar is.

In dit voorbeeld wordt een query uitgevoerd op de opgegeven tabel en worden de eerste vijf rijen geretourneerd. In dit voorbeeld wordt de SPARK_REMOTE omgevingsvariabele gebruikt voor verificatie.

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")
  }
}

Aanvullende bronnen

Databricks biedt aanvullende voorbeeldtoepassingen die laten zien hoe u Databricks Connect gebruikt in de GitHub-opslagplaats van Databricks Connect, waaronder:

• Een eenvoudige ETL-toepassing
• Visualisaties van grafieken met JFreeChart