„INSERT INTO ...” z SparkSQL HiveContext

Question

Próbuję uruchomić instrukcji INSERT z moim HiveContext, tak:

hiveContext.sql('insert into my_table (id, score) values (1, 10)') 

1.5.2 Spark SQL Documentation nie jednoznacznie stwierdzić, czy ten jest obsługiwany czy nie, chociaż obsługuje "dynamiczne wstawianie partycji".

Prowadzi to do śledzenia stosu jak

AnalysisException: 
Unsupported language features in query: insert into my_table (id, score) values (1, 10) 
TOK_QUERY 0, 0,20, 0 
    TOK_FROM 0, -1,20, 0 
    TOK_VIRTUAL_TABLE 0, -1,20, 0 
     TOK_VIRTUAL_TABREF 0, -1,-1, 0 
     TOK_ANONYMOUS 0, -1,-1, 0 
     TOK_VALUES_TABLE 1, 13,20, 41 
     TOK_VALUE_ROW 1, 15,20, 41 
      1 1, 16,16, 41 
      10 1, 19,19, 44 
    TOK_INSERT 1, 0,-1, 12 
    TOK_INSERT_INTO 1, 0,11, 12 
     TOK_TAB 1, 4,4, 12 
     TOK_TABNAME 1, 4,4, 12 
      my_table 1, 4,4, 12 
     TOK_TABCOLNAME 1, 7,10, 22 
     id 1, 7,7, 22 
     score 1, 10,10, 26 
    TOK_SELECT 0, -1,-1, 0 
     TOK_SELEXPR 0, -1,-1, 0 
     TOK_ALLCOLREF 0, -1,-1, 0 

scala.NotImplementedError: No parse rules for: 
TOK_VIRTUAL_TABLE 0, -1,20, 0 
    TOK_VIRTUAL_TABREF 0, -1,-1, 0 
    TOK_ANONYMOUS 0, -1,-1, 0 
    TOK_VALUES_TABLE 1, 13,20, 41 
    TOK_VALUE_ROW 1, 15,20, 41 
     1 1, 16,16, 41 
     10 1, 19,19, 44 

Czy jest jakiś inny sposób, aby wstawić do tabeli Hive że jest obsługiwane?

Answer 1

Dane mogą być dołączane do tabeli gałęzi przy użyciu trybu append w DataFrameWriter.

data = hc.sql("select 1 as id, 10 as score") 
data.write.mode("append").saveAsTable("my_table") 

Daje taki sam wynik, jak wstawka.

Answer 2

Miałem ten sam problem (Spark 1.5.1) i wypróbowałem różne wersje.

Podane

sqlContext.sql("create table my_table(id int, score int)") 

Jedyne wersje, które pracowały wyglądał następująco:

sqlContext.sql("insert into table my_table select t.* from (select 1, 10) t") 
sqlContext.sql("insert into  my_table select t.* from (select 2, 20) t") 

Answer 3

spróbować hiveContext.sql("insert into table my_table select 1, 10") jeśli nie zmienisz tryb dynamiczny partycji do nonstrict, trzeba to zrobić hiveCtx.setConf("hive.exec.dynamic.partition.mode", "nonstrict")

Answer 4

Przyjęta odpowiedź saveAsTable nie powiedzie się dla mnie z AnalysisException (Nie rozumiem dlaczego). To, co działa dla mnie, to:

data = hc.sql("select 1 as id, 10 as score") 
data.write.mode("append").insertInto("my_table") 

Używam Sparka v2.1.0.

Answer 5

Podjęto próbę wykonania czynności, której format pliku danych nie jest możliwy, stąd wyjątek Unsupported language features in query.

Wiele formatów plików danych zapisuje się jednokrotnie i nie obsługuje operacji ACID.

Apache ORC obsługuje operację ACID, jeśli jej potrzebujesz.

Zamiast tego możesz użyć partycji do podzielenia danych na foldery (/ data/rok = 2017/miesiąc = 10 ....), tutaj możesz dołączyć/wstawić dane do swojego jeziora danych.

Answer 6

Po raz pierwszy to zrobić

$data.write.mode("append").saveAsTable("my_table") 

należy wymienić "append" z "overwrite", Następnie można użyć "append".