Можно ли сохранить DataFrame
в Spark прямо в Hive?
Я попытался преобразовать DataFrame
в Rdd
текстовый файл, а затем сохранить его и загрузить в куст. Но мне интересно, могу ли я напрямую сохранить dataframe
в улей
Можно ли сохранить DataFrame
в Spark прямо в Hive?
Я попытался преобразовать DataFrame
в Rdd
текстовый файл, а затем сохранить его и загрузить в куст. Но мне интересно, могу ли я напрямую сохранить dataframe
в улей
Вы можете создать временную таблицу в памяти и сохранить их в таблице кустов с помощью sqlContext.
Допустим, ваш фрейм данных - myDf. Вы можете создать одну временную таблицу, используя,
myDf.createOrReplaceTempView("mytempTable")
Затем вы можете использовать простой оператор hive для создания таблицы и выгрузки данных из вашей временной таблицы.
sqlContext.sql("create table mytable as select * from mytempTable");
temporary
стол соhive
столом? При этомshow tables
включаются толькоhive
таблицы для моейspark 2.3.0
установкиИспользуйте
DataFrameWriter.saveAsTable
. (df.write.saveAsTable(...)
) См. Руководство по Spark SQL и DataFrame .источник
df.write().saveAsTable(tableName)
также будет записывать потоковые данные в таблицу?Я не вижу
df.write.saveAsTable(...)
устаревших в документации Spark 2.0. У нас это сработало на Amazon EMR. Мы прекрасно могли считывать данные из S3 в фрейм данных, обрабатывать их, создавать таблицу из результатов и читать их с помощью MicroStrategy. Ответ Vinays также сработал.источник
вам нужно иметь / создать HiveContext
import org.apache.spark.sql.hive.HiveContext; HiveContext sqlContext = new org.apache.spark.sql.hive.HiveContext(sc.sc());
Затем напрямую сохраните фрейм данных или выберите столбцы для хранения в виде таблицы улья.
df - это фрейм данных
df.write().mode("overwrite").saveAsTable("schemaName.tableName");
или же
df.select(df.col("col1"),df.col("col2"), df.col("col3")) .write().mode("overwrite").saveAsTable("schemaName.tableName");
или же
df.write().mode(SaveMode.Overwrite).saveAsTable("dbName.tableName");
SaveModes: Append / Ignore / Overwrite / ErrorIfExists.
Я добавил сюда определение HiveContext из документации Spark,
В дополнение к базовому SQLContext вы также можете создать HiveContext, который обеспечивает расширенный набор функций, предоставляемых базовым SQLContext. Дополнительные функции включают возможность писать запросы с использованием более полного анализатора HiveQL, доступ к пользовательским функциям Hive и возможность чтения данных из таблиц Hive. Чтобы использовать HiveContext, вам не нужно иметь существующую настройку Hive, и все источники данных, доступные для SQLContext, по-прежнему доступны. HiveContext упаковывается отдельно, чтобы избежать включения всех зависимостей Hive в сборку Spark по умолчанию.
в Spark версии 1.6.2 использование "dbName.tableName" дает такую ошибку:
источник
df.write().mode...
необходимо изменить наdf.write.mode...
Сохранение в Hive - это просто вопрос использования
write()
метода вашего SQLContext:См. Https://spark.apache.org/docs/2.1.0/api/java/org/apache/spark/sql/DataFrameWriter.html#saveAsTable(java.lang.String)
Начиная с Spark 2.2: используйте DataSet вместо DataFrame.
источник
From Spark 2.2: use DataSet instead DataFrame.
Извините, что написал поздно, но я не вижу принятого ответа.
df.write().saveAsTable
будет выбрасыватьAnalysisException
и не совместим со столом HIVE.Сохранение DF как
df.write().format("hive")
должно помочь!Однако, если это не сработает, то, судя по предыдущим комментариям и ответам, на мой взгляд, это лучшее решение (хотя открыто для предложений).
Лучший подход - явно создать таблицу HIVE (включая таблицу PARTITIONED),
def createHiveTable: Unit ={ spark.sql("CREATE TABLE $hive_table_name($fields) " + "PARTITIONED BY ($partition_column String) STORED AS $StorageType") }
сохранить DF как временную таблицу,
df.createOrReplaceTempView("$tempTableName")
и вставить в таблицу РАЗДЕЛЕННЫЙ УЛЕЙ:
spark.sql("insert into table default.$hive_table_name PARTITION($partition_column) select * from $tempTableName") spark.sql("select * from default.$hive_table_name").show(1000,false)
Offcourse последний столбец в DF будет PARTITION COLUMN таким образом , создать Hive таблицы соответственно!
Прокомментируйте, если это работает! или нет.
--ОБНОВИТЬ--
df.write() .partitionBy("$partition_column") .format("hive") .mode(SaveMode.append) .saveAsTable($new_table_name_to_be_created_in_hive) //Table should not exist OR should be a PARTITIONED table in HIVE
источник
Вот версия PySpark для создания таблицы Hive из паркетного файла. Возможно, вы сгенерировали файлы Parquet с использованием предполагаемой схемы и теперь хотите отправить определение в хранилище метаданных Hive. Вы также можете отправить определение в систему, например AWS Glue или AWS Athena, а не только в хранилище метаданных Hive. Здесь я использую spark.sql для создания / создания постоянной таблицы.
# Location where my parquet files are present. df = spark.read.parquet("s3://my-location/data/") cols = df.dtypes buf = [] buf.append('CREATE EXTERNAL TABLE test123 (') keyanddatatypes = df.dtypes sizeof = len(df.dtypes) print ("size----------",sizeof) count=1; for eachvalue in keyanddatatypes: print count,sizeof,eachvalue if count == sizeof: total = str(eachvalue[0])+str(' ')+str(eachvalue[1]) else: total = str(eachvalue[0]) + str(' ') + str(eachvalue[1]) + str(',') buf.append(total) count = count + 1 buf.append(' )') buf.append(' STORED as parquet ') buf.append("LOCATION") buf.append("'") buf.append('s3://my-location/data/') buf.append("'") buf.append("'") ##partition by pt tabledef = ''.join(buf) print "---------print definition ---------" print tabledef ## create a table using spark.sql. Assuming you are using spark 2.1+ spark.sql(tabledef);
источник
Для внешних таблиц Hive я использую эту функцию в PySpark:
def save_table(sparkSession, dataframe, database, table_name, save_format="PARQUET"): print("Saving result in {}.{}".format(database, table_name)) output_schema = "," \ .join(["{} {}".format(x.name.lower(), x.dataType) for x in list(dataframe.schema)]) \ .replace("StringType", "STRING") \ .replace("IntegerType", "INT") \ .replace("DateType", "DATE") \ .replace("LongType", "INT") \ .replace("TimestampType", "INT") \ .replace("BooleanType", "BOOLEAN") \ .replace("FloatType", "FLOAT")\ .replace("DoubleType","FLOAT") output_schema = re.sub(r'DecimalType[(][0-9]+,[0-9]+[)]', 'FLOAT', output_schema) sparkSession.sql("DROP TABLE IF EXISTS {}.{}".format(database, table_name)) query = "CREATE EXTERNAL TABLE IF NOT EXISTS {}.{} ({}) STORED AS {} LOCATION '/user/hive/{}/{}'" \ .format(database, table_name, output_schema, save_format, database, table_name) sparkSession.sql(query) dataframe.write.insertInto('{}.{}'.format(database, table_name),overwrite = True)
источник
В моем случае это нормально работает:
from pyspark_llap import HiveWarehouseSession hive = HiveWarehouseSession.session(spark).build() hive.setDatabase("DatabaseName") df = spark.read.format("csv").option("Header",True).load("/user/csvlocation.csv") df.write.format(HiveWarehouseSession().HIVE_WAREHOUSE_CONNECTOR).option("table",<tablename>).save()
Готово!!
Вы можете читать данные, позволяя указать как «Сотрудник»
hive.executeQuery("select * from Employee").show()
Для получения дополнительных сведений используйте этот URL-адрес: https://docs.cloudera.com/HDPDocuments/HDP3/HDP-3.1.5/integrating-hive/content/hive-read-write-operations.html
источник
Вы можете использовать библиотеку искры Hortonworks следующим образом
import com.hortonworks.hwc.HiveWarehouseSession df.write .format("com.hortonworks.spark.sql.hive.llap.HiveWarehouseConnector") .mode("append") .option("table", "myDatabase.myTable") .save()
источник
val df = ... val schemaStr = df.schema.toDDL # This gives the columns spark.sql(s"""create table hive_table ( ${schemaStr})""") //Now write the dataframe to the table df.write.saveAsTable("hive_table")
hive_table
будет создан в пространстве по умолчанию, поскольку мы не предоставили никакой базы данных по адресуspark.sql()
.stg.hive_table
можно использовать для созданияhive_table
вstg
базе данных.источник