Пример использования сеанса Databricks Spark

В более новой версии Apache Spark SparkSession можно использовать без инициализации.

Example: spark.read.parquet("/pathToParquet")

Но рекомендуется следовать шаблону компоновщика и создавать новый сеанс Spark перед использованием переменной spark.

Шаблон Строителя

import org.apache.spark.sql.SparkSession
val spark = SparkSession
 .builder()
 .appName("yourAppName")
 .enableHiveSupport()
 .getOrCreate()
 
spark.read.parquet("/pathToParquet")

Первое преимущество заключается в том, что когда несколько заданий выполняются одновременно, AppName поможет различать журналы. Кроме того, это поможет в уведомлении Microsoft Teams.

Помимо этого, Spark Session также позволяет использовать настраиваемые атрибуты, которые будут очень удобны для обмена данными между Scala и PySpark.

Пример:

val spark = SparkSession
   .builder()
   .appName("yourAppName")
   .enableHiveSupport()
   .getOrCreate()
   
spark.conf.set("my_startDt","2021-05-06")
spark.conf.set("my_endDt","2021-05-06")

Теперь, что происходит, сеанс Spark имеет 2 настраиваемые переменные, которые могут быть прочитаны PySpark, а также Scala в любой момент времени.

%scala
val appName = spark.conf.get("spark.app.name")
val startDt = spark.conf.get("my_startDt")
val endDt = spark.conf.get("my_endDt")
print(f"${startDt},${endDt}")
%python
startDt = spark.conf.get("my_startDt")
endDt = spark.conf.get("my_endDt")
print(f"{startDt}, {endDt}")

Используя метод getAll, можно прочитать и отобразить все значения конфигурации искры за один раз.

%scala
val configMap = spark.conf.getAll
configMap.foreach(println)

Метод 2. Передача переменной / значения из Scala в PySpark

Создайте временное представление из фрейма данных scala.
Считайте значения столбцов из временного представления с помощью PySpark.

%scala
var dfPassData = Seq("2021-05-06").toDF("my_startDt")
                .withColumn("my_endDt",lit("2021-05-06"))
                
dfPassData.createOrReplaceTempView("tmpPassData")

%python
pystartDate=table("tmpPassData").head().my_startDt
pyendDate=table("tmpPassData").head().my_endDt

Больше контента на plainenglish.io