我们正在尝试为配置单元运行insert sql,数据来自spark中的Dataframe。使用的会话包含所有内容。
有两个问题:
问题)尽管我们在foreach循环中创建了会话,但在尝试两者时插入仍然失败
1) Dataframe
2) 直Sparksql
下面是代码(spark sql方法):
import java.time.Instant
import org.apache.spark.sql.{DataFrame, Row, types}
import org.apache.spark.sql.functions.{current_timestamp, first, isnull, lit, max}
import org.apache.spark.sql.types.{StringType, StructField, StructType, TimestampType}
import scala.collection.mutable.ListBuffer
class Controller extends DatabaseServices
with Loggers {
val session = createSparkSession(ConfigFactory.load().getString("local.common.spark.app.name"))
val producer = session.sparkContext.broadcast(KafkaWrapper())
def doIt(TranIDs: DataFrame): Unit = {
import session.sqlContext.implicits._
val TranID = TranIDs
.withColumnRenamed("TranID", "REFERENCE_TranID")
.select($"REFERENCE_TranID")
.union(session.table(BANK_ROLLBACK_TXN_PRODUCER_LOG_VIEW)
.withColumnRenamed("TranID", "REFERENCE_TranID")
.select($"REFERENCE_TranID"))
.where($"REFERENCE_TranID".isNotNull)
if (TranID.count() == 0) {
throw new Exception("No rows.")
}
val core = session
.table(BANK_TRANS_MASTER_CORE)
.withColumnRenamed("TranID", "MASTER_REFERENCE_TranID")
.withColumnRenamed("CLIENTID", "REF_CLIENT_ID")
.withColumnRenamed("SUBCLIENTID", "REF_SUBCLIENT_ID")
.select($"MASTER_REFERENCE_TranID",
$"TranIDDATE")
.join(TranID, TranID.col("REFERENCE_TranID") === $"MASTER_REFERENCE_TranID")
val ref = session
.table(BANK_RBI_REF_CLIENT)
.select($"CLIENTID", $"SUBCLIENTID", $"FLAGTRE")
.join(core, $"CLIENTID" === core.col("REF_CLIENT_ID")
&& $"SUBCLIENTID" === core.col("REF_SUBCLIENT_ID")
val details = session
.table(BANK_TRANS_MASTER_DETAILS)
.select($"TranID",
$"REALFRAUD",
$"REALFRAUDDATEBAE",
$"REALFRAUDYYYYMMDD"
)
.join(ref, ref.col("MASTER_REFERENCE_TranID") === $"TranID"
&& $"REALFRAUD" === lit("Y"))
.where($"TranID".isNotNull
&& $"TranIDDATE".isNotNull)
.groupBy($"TranID")
.agg(first($"TranID").as("TranID"),
first(core("TranIDDATE")).cast("String").as("TranIDDATE"),
max($"REALFRAUDDATEBAE").as("REALFRAUDDATEBAE"),
max($"REALFRAUDYYYYMMDD").as("REALFRAUDYYYYMMDD"),
first($"REALFRAUD").as("REALFRAUD"),
first($"ABA").as("ABA"))
details.foreach(row => {
import scala.collection.JavaConversions._
val transaction = TxUpdate.newBuilder().setTranID(row.getAs("TranID").toString)
.setTranIDDATE(row.getAs("TranIDDATE").toString)
.setAttributes(ListBuffer(
Attribute.newBuilder.setKey("REALFRAUD").setValue(if (row.getAs("REALFRAUD") != null) row.getAs("REALFRAUD").toString else null).build(),
Attribute.newBuilder.setKey("REALFRAUDDATEBAE").setValue(if (row.getAs("REALFRAUDDATEBAE") != null) if (row.getAs("REALFRAUDDATEBAE") != null) row.getAs("REALFRAUDDATEBAE").toString else null else null).build(),
Attribute.newBuilder.setKey("REALFRAUDYYYYMMDD").setValue(if (row.getAs("REALFRAUDYYYYMMDD") != null) row.getAs("REALFRAUDYYYYMMDD").toString else null).build(),
Attribute.newBuilder.setKey("ABA").setValue(if (row.getAs("ABA") != null) row.getAs("ABA").toString else null).build(),
.build()
if (producer.value.sendSync(ConfigFactory.load().getString("local.common.kafka.rollbackKafkaTopicName"),
transaction.getTranID.toString,
transaction)) {
session.sqlContext.sql("insert into " + BANK_ROLLBACK_TXN_PRODUCER_LOG + "(TranID, when_loaded, status) values('" + transaction.getTranID.toString + "', 'current_timestamp()', 'S')")
} else {
session.sqlContext.sql("insert into " + BANK_ROLLBACK_TXN_PRODUCER_LOG + "(TranID, when_loaded, status) values('" + transaction.getTranID.toString + "', 'current_timestamp()', 'F')")
}
})
}
}
2条答案
按热度按时间gwo2fgha1#
这里的错误不清楚。
在高层,您可以使用在spark中启用hivecontext的方法,然后使用append选项直接持久化到配置单元表中。这将比执行插入操作快得多。流程如下:
第0步-所有这些都必须在一个星火会议发生。不需要为每个插入创建多个会话。在某种程度上这是没有意义的。答。创建一个包含配置单元的基础表的列的数据框。b。在spark处理期间,Dataframe最终将其数据持久化到hive中。c。使用append选项启动dataframe saveastable
插入Hive
希望这有助于理解如何处理这个问题。
lskq00tm2#
我们使用了df.write.insertinto和append选项,这会导致错误。终于成功了。