Как ссылаться на широковещательную переменную в Spark DataFrameSQL

Question

У меня есть следующий SparkSQL:

val resultDf = spark.sql("SELECT name, phone, country FROM users")

Я бы хотел отфильтровать возвращенные записи по странам, которые присутствуют в следующей коллекции:

val countries = Seq("Italy", "France", "United States", "Poland", "Spain")

Например, я могу создать широковещательную переменную на основе коллекции:

val countriesBroadcast = sc.broadcast(countries)

но возможно ли (и если да - как?) Использовать переменную countriesBroadcast внутри моего SQL-запроса?

Answer 1

В API фрейма данных spark мы можем транслировать всю таблицу, можно объединить ее с целевой таблицей, чтобы получить желаемый результат. Вот пример кода.

Импорт

import org.apache.spark.sql.functions.broadcast

код

val spark = SparkSession.builder().master("local").getOrCreate()

val df = spark.read.option("header", true).csv("data/user.txt")
df.createOrReplaceTempView("users")
val countries = Seq("Italy", "France", "United States", "Poland", "Spain")
import spark.implicits._
spark.sparkContext.parallelize(countries, 1).toDF("country").createOrReplaceTempView("countries")
broadcast(spark.table("countries")).join(spark.table("users"), "country").show()

содержимое файла «data / user.txt»

 name,phone,country
 a,123,India
 b,234,Italy
 c,526,France
 d,765,India

Вывод кода:

+-------+----+-----+
|country|name|phone|
+-------+----+-----+
|  Italy|   b|  234|
| France|   c|  526|
+-------+----+-----+

Примечание: код, протестированный с Spark 2.2 и Scala 2.11

Answer 2

Это невозможно, за исключением UserDefinedFunctions, UserDefinedAggregateFunctions и Aggregators (т.е. не декларативный код).

Для использования широковещания с DataFrame / SQL API вы должны использовать DataFrames и использовать подсказку широковещания - Spark SQL широковещательное хеш-соединение .