Блог

Я пытаюсь извлечь данные из таблицы с помощью spark jdbc read.

 var df = spark.read.format("csv").option("header", "true").option("sep", ",").load("abfss://test@a.dfs.core.windows.net/data/raw")

import org.apache.spark.sql
df = df.withColumn("CustomerID", $"CustomerID".cast(sql.types.IntegerType)).withColumn("LAST_UPDATED", $"LAST_UPDATED".cast(sql.types.DateType))
val customerIds = df.select("CustomerID").collect()
val inCondition = "('"   customerIds.mkString("','")   "')"
val jdbcData = spark.read.format("jdbc").options(Map("url" -> jdbcUrl, "dbtable" -> "(SELECT * FROM [dbo].[SinkCustomers] WHERE CustomerID IN ("  inCondition  ")) AS C")).load()
 

Но результат таков,

  found   : String
 required: (?, ?)
 .options(Map("url" -> jdbcUrl, "dbtable" -> "(SELECT * FROM [dbo].[SinkCustomers] WHERE CustomerID IN ("  inCondition  ")) AS C"))
 

Выдает эту ошибку, и я также попытался ввести жестко закодированное значение int. Но все равно терпит неудачу. Но если я запущу (SELECT * FROM [dbo].[SinkCustomers] WHERE CustomerID IN (12323)) AS C") , это сработает хорошо. Неверно ли то, как я выполняю конкатенацию строк, или это проблема с идентификаторами клиентов? Что я могу сделать, чтобы решить эту проблему?

Используйте интерполяцию строк s".." , чтобы правильно построить свой запрос.

    spark
      .read
      .format("jdbc")
      .options(Map("url" -> jdbcUrl, "dbtable" -> s"(SELECT * FROM [dbo].[SinkCustomers] WHERE CustomerID IN ($inCondition)) AS C"))
      .load()
 

Вне темы: я бы избегал var df и использовал несколько val переменных