INSERT HIVE SQL в цикле в Spark

Question

Мы пытаемся запустить INSERT SQL для HIVE с данными, поступающими из кадра данных в Spark.В используемом сеансе есть все и т.д.) Фреймы данных

2) Прямой Spark SQL

Ниже приведен код (подход 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')")
      }

    })

  }
}

Answer 1

Мы использовали df.write.insertInto с опцией Append, которая выдавала ошибки.Наконец-то сделал это для работы.

Answer 2

Ошибка здесь не ясна.

На высоком уровне вы можете использовать подход включения hivecontext в Spark, а затем напрямую сохранять его в таблице Hive, используя опцию добавления.Это будет намного быстрее, чем операция вставки.Поток будет примерно таким:

шаг 0 - все это должно произойти за один сеанс искры.Вам не нужно создавать несколько сессий для каждой вставки.В каком-то смысле это бессмысленно делать.а.Создайте фрейм данных, имеющий столбцы базовой таблицы Hive.б.Во время искровой обработки фрейм данных получает данные для сохранения в конечном итоге в Hive.с.Инициируйте сохранение данных в фрейме с опцией добавления

Вставить в улей

Надеюсь, это поможет понять, как вам нужно решить эту проблему.