Есть вот такой пример с кодом который якобы работает
https://blog.knoldus.com/2017/06/02/dea ... -redshift/
Когда я запускаю свой вариант - все грустно падает в джавовском форматтере.
В частности вит в этой строке
val deltaQuery = postActionsQuery.format(targetTable, sourceTable, stagingTable)
Не пойму как такое может быть что у автора работает с "$", а у меня нет ?
Code: Select all
def writeIntoTable(sourceDF: DataFrame, jdbcURL: String, tempS3Dir: String, targetTable: String): Unit = {
val stagingTable = s"{targetTable}Staging"
//println(stagingTable.toString)
val sourceTable = s"{targetTable}Source"
//println(sourceTable.toString)
val insertUpdatedRecordsInStaging = """Insert into %$3s Select %2$s.* from %2$s join %1$s on %1$s.hash = %2$s.hash;"""
//println(insertUpdatedRecordsInStaging.toString)
val insertUpdatedRecordsInTarget =
"""Delete from %1$s using %3$s where %1$s.hash = %3$s.hash;
|Insert into %1$s Select * from %3$s;""".stripMargin
//println(insertUpdatedRecordsInTarget.toString)
val insertNewRecordsInTarget =
"""Delete from %2$s using %1$s where %2$s.hash = %1$s.hash;
|Insert into %1$s Select * from %2$s;""".stripMargin
//println(insertNewRecordsInTarget.toString)
val postActionsQuery = s"""${insertUpdatedRecordsInStaging}begin transaction;${insertUpdatedRecordsInTarget} ${insertNewRecordsInTarget}end transaction;DROP tables $sourceTable,$stagingTable;"""
println(postActionsQuery)
val deltaQuery = postActionsQuery.format(targetTable, sourceTable, stagingTable)2017-10-27 13:20:26,755 INFO [main] - (o.a.s.sql.execution.SparkSqlParser:54) - Parsing command: jobInstanceId as job_instance_id
Exception in thread "main" java.util.UnknownFormatConversionException: Conversion = '$'
Insert into %$3s Select %2$s.* from %2$s join %1$s on %1$s.hash = %2$s.hash;begin transaction;Delete from %1$s using %3$s where %1$s.hash = %3$s.hash;
at java.util.Formatter.checkText(Formatter.java:2579)
at java.util.Formatter.parse(Formatter.java:2555)
Insert into %1$s Select * from %3$s; Delete from %2$s using %1$s where %2$s.hash = %1$s.hash;
at java.util.Formatter.format(Formatter.java:2501)
at java.util.Formatter.format(Formatter.java:2455)
Insert into %1$s Select * from %2$s;end transaction;DROP tables {targetTable}Source,{targetTable}Staging;
at java.lang.String.format(String.java:2940)
at scala.collection.immutable.StringLike$class.format(StringLike.scala:318)
at scala.collection.immutable.StringOps.format(StringOps.scala:29)
at ......spark.redshift.RedshiftMergeTest$.writeIntoTable(RedshiftMergeTest.scala:111)
at ......spark.redshift.RedshiftMergeTest$.main(RedshiftMergeTest.scala:76)
at .......spark.redshift.RedshiftMergeTest.main(RedshiftMergeTest.scala)
.
