Передать значения даты из кадра данных для запроса в Spark / Scala

Question

У меня есть датафрейм со столбцом даты и значениями, как показано ниже.

df.show()
+----+----------+
|name|       dob|
+----+----------+
| Jon|2001-04-15|
| Ben|2002-03-01|
+----+----------+

Теперь мне нужно запросить таблицу в кусте, в которой есть "dob" из вышеупомянутого фрейма данных (оба 2001-04-15 , 2002-03-01). Так что мне нужно передать значения в столбце dob в качестве параметра в мой запрос улья.

Я попытался собрать значения в переменную, как показано ниже, которая дает мне массив строк.

val dobRead = df.select("updt_d").distinct().as[String].collect()
dobRead: Array[String] = Array(2001-04-15, 2002-03-01)

Однако, когда я пытаюсь перейти к запросу, я вижу, что он не подставляется должным образом и выдает ошибку.

val tableRead = hive.executeQuery(s"select emp_name,emp_no,martial_status from <<table_name>> where dateOfBirth in ($dobRead)")

org.apache.hadoop.hive.ql.metadata.HiveException: Failed to compile query: org.apache.hadoop.hive.ql.parse.ParseException: line 1:480 cannot recognize input near '(' '[' 'Ljava' in expression specification

Не могли бы вы помочь мне, как передать значения даты на запрос в спарк.

Answer 1

Вы можете собрать даты следующим образом ( Row.getAs ):

val rows: Array[Row] = df.select("updt_d").distinct().collect()
val dates: Array[String] = rows.map(_.getAs[String](0))

И затем построить запрос:

val hql: String = s"select ... where dateOfBirth in (${
  dates.map(d => s"'${d}'").mkString(", ")
})"

Вариант 2

Если число дат в первом DataFrame слишком велико, вы должны использовать операции соединения вместо того, чтобы собирать их в драйвер.

Сначала загрузите каждую таблицу как DataFrames (я назову их dfEmp и dfDates). Затем вы можете объединить поля даты для фильтрации, либо используя стандартное inner объединение плюс фильтрация пустых полей, либо непосредственно left_semi объединение:

val dfEmp = hiveContext.table("EmpTable")
val dfEmpFiltered = dfEmp.join(dfDates,
  col("dateOfBirth") === col("updt_d"), "left_semi")