Не удалось подключить SparkSession к Hive - PullRequest
Новая
       8

Не удалось подключить SparkSession к Hive

0 голосов
/ 13 октября 2018

Я получаю данные из файлов .csv, а затем регистрирую df как TempView.После этого я пытаюсь записать содержимое из TempView в таблицы Hive.Однако я получаю сообщение об ошибке, что база данных не найдена.Но у меня есть база данных в улье.

Ниже приведен код: 

import org.apache.spark.sql.types._
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.col


object read_data {


  def main (args:Array[String]) {

    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("Spark SQL basic example")
      .config("spark.executor.memory", "1g")
      .config("hive.metastore.warehouse.dir", "user/hive/warehouse")
      .enableHiveSupport()
      .getOrCreate()

    ///val hiveContext =  new org.apache.spark.sql.hive.HiveContext(spark);


    val customSchema = StructType(Array(
      StructField("order_id", IntegerType, true),
      StructField("parent_order_uuid", StringType, true),
      StructField("company", StringType, true),
      StructField("country_id", IntegerType, true)))

    val readogp = spark.read.format("csv")
      .option("header", "false")
      .schema(customSchema)
      .load("/home/cloudera/Desktop/ogp_csv.csv")

   // readogp.createOrReplaceTempView("OGP_INTL")

    val read_country = spark.read.format("csv")
      .option("header", "true")
      .load("/home/cloudera/Desktop/country.csv")


    //read_country.createOrReplaceTempView("Dim_Country")
      println("************************************************************************")


    val rept = readogp.join(read_country, readogp("country_id") === read_country("country_id"),"inner")

    val final_rp = rept.select(col("order_id"), col("country_iso_code_2"))
      .groupBy("country_iso_code_2")
      .sum("order_id")
      .orderBy("country_iso_code_2")

      final_rp.createOrReplaceTempView("FINAL_REPORT_OGP")


    spark.sql("use ods")
    spark.sql("""insert into ods.final_ogp
       select * from FINAL_REPORT_OGP""")

  }

}

Журнал ошибок: 

18/10/13 13:01:56 INFO HiveMetaStore: 0: get_database: ods
18/10/13 13:01:56 INFO audit: ugi=cloudera  ip=unknown-ip-addr  cmd=get_database: ods   
18/10/13 13:01:56 WARN ObjectStore: Failed to get database ods, returning NoSuchObjectException
Exception in thread "main" org.apache.spark.sql.catalyst.analysis.NoSuchDatabaseException: Database 'ods' not found;

Коэффициенты базы данных существуют в Hive, и поэтомуделает таблицу.Я сделал ошибку в коде?

Ответы [ 2 ]

0 голосов
/ 15 октября 2018

Программа определенно не может подключиться к хранилищу Hive, возможно, вы указали неверный путь к хранилищу.Пожалуйста, исправьте путь: "пользователь / улей / склад" Я думаю, это будет: " / пользователь / улей / склад " (Абсолютный путь)

Также еще одна проверка: вы получите эту проблему, если вы запускаете свою программу из Intellij (по сути, любой редактор).Если да, добавьте hive-site.xml в ресурсы вашего проекта: / src / main / resources Также вам необходимо добавить jar, например: Jive Jive Hive, коннектор mySQL (если метастор MySQL).

Hopeэто помогает.

0 голосов
/ 14 октября 2018

Первое, что я могу проверить, это значение spark conf spark.sql.catalogImplementation, оно должно быть установлено на hive .

Это можно найти в пользовательском интерфейсе Spark (в локальном режиме он должен быть доступен по умолчанию http://localhost:4040).
При вызове enableHiveSupport при создании SparkSession устанавливается указанное вышеПосле этого conf и Spark используют HiveCatalog.
Если значение conf равно в памяти , то используется каталог Spark в памяти.

Я вижу следующую реализацию для enableHiveSupport, 

def enableHiveSupport(): Builder = synchronized {
      if (hiveClassesArePresent) {
        config(CATALOG_IMPLEMENTATION.key, "hive")
      } else {
        throw new IllegalArgumentException(
          "Unable to instantiate SparkSession with Hive support because " +
            "Hive classes are not found.")
      }
    }




  private[spark] def hiveClassesArePresent: Boolean = {
    try {
      Utils.classForName(HIVE_SESSION_STATE_BUILDER_CLASS_NAME)
      Utils.classForName("org.apache.hadoop.hive.conf.HiveConf")
      true
    } catch {
      case _: ClassNotFoundException | _: NoClassDefFoundError => false
    }
  }

Из приведенных выше фрагментов может быть, что путь к классу драйвера неправильно установлен для включения org.apache.hadoop.hive.conf.HiveConf

...