内存分配问题蜂巢表

我试图挽救一个Spark数据帧的蜂巢表（木地板），在pySpark .saveAsTable()，但要像下面内存问题运行在：内存分配问题蜂巢表

org.apache.hadoop.hive.ql.metadata.HiveException: parquet.hadoop.MemoryManager$1: 
New Memory allocation 1034931 bytes is smaller than the minimum allocation size of 1048576 bytes.

第一个号码（1034931）通常在不同的运行中保持不变。我认识到第二个数字（1048576）是1024^2，但我不知道这里的含义。

我一直在使用完全相同的技术用于其他一些项目（具有更大的DataFrames），并且它没有问题。在这里，我基本上是复制粘贴了进程和配置的结构，但却遇到了内存问题！它一定是我失踪的小事。

的Spark数据框（我们称之为sdf）具有结构（〜10列〜300K行，但可能是更多，如果该程序运行正常）：

+----------+----------+----------+---------------+---------------+ 
| col_a_str| col_b_num| col_c_num|partition_d_str|partition_e_str| 
+----------+----------+----------+---------------+---------------+ 
|val_a1_str|val_b1_num|val_c1_num|  val_d1_str|  val_e1_str| 
|val_a2_str|val_b2_num|val_c2_num|  val_d2_str|  val_e2_str| 
|  ...|  ...|  ...|   ...|   ...| 
+----------+----------+----------+---------------+---------------+

Hive的表是这样创建的：

sqlContext.sql(""" 
        CREATE TABLE IF NOT EXISTS my_hive_table (
         col_a_str string, 
         col_b_num double, 
         col_c_num double 
        ) 
        PARTITIONED BY (partition_d_str string, 
            partition_e_str string) 
        STORED AS PARQUETFILE 
       """)

在插入数据到该表中的尝试是使用下面的命令：

sdf.write \ 
    .mode('append') \ 
    .partitionBy('partition_d_str', 'partition_e_str') \ 
    .saveAsTable('my_hive_table')

星火/蜂巢结构是这样的：

spark_conf = pyspark.SparkConf() 
spark_conf.setAppName('my_project') 

spark_conf.set('spark.executor.memory', '16g') 
spark_conf.set('spark.python.worker.memory', '8g') 
spark_conf.set('spark.yarn.executor.memoryOverhead', '15000') 
spark_conf.set('spark.dynamicAllocation.maxExecutors', '64') 
spark_conf.set('spark.executor.cores', '4') 

sc = pyspark.SparkContext(conf=spark_conf) 

sqlContext = pyspark.sql.HiveContext(sc) 
sqlContext.setConf('hive.exec.dynamic.partition', 'true') 
sqlContext.setConf('hive.exec.max.dynamic.partitions', '5000') 
sqlContext.setConf('hive.exec.dynamic.partition.mode', 'nonstrict') 
sqlContext.setConf('hive.exec.compress.output', 'true')

我曾试图改变.partitionBy('partition_d_str', 'partition_e_str')到.partitionBy(['partition_d_str', 'partition_e_str'])，增加内存，分割数据帧，以更小的块，重新创建表和数据帧，但似乎没有任何工作。我无法在网上找到任何解决方案。什么会导致内存错误（我不完全理解它来自哪里），以及如何更改我的代码以写入Hive表？谢谢。

来源

2017-05-17 vk1011

实木复合地板的最小页面大小（即最小读/写单位）由属性parquet.page.size定义，默认为1048576.其试图写入的数据可能低于此阈值。那就是为什么投掷错误可能是？这只是我的猜测... [检查了这一点]（https://github.com/Parquet/parquet-mr/blob/fa8957d7939b59e8d391fa17000b34e865de015d/parquet-hadoop/src/main/java/parquet/hadoop/ParquetOutputFormat.java# L64） – Pushkr

感谢您的链接。通过玩'parquet.page.size'和'parquet.block.size'配置，以及通过乘以我的数据的大小，但没有运气尝试了你的建议。相同的错误:( – vk1011

事实证明，我正在用可空字段进行分区，抛出.saveAsTable()。

from pyspark.sql.types import * 

# Define schema 
my_schema = StructType(
        [StructField('col_a_str', StringType(), False), 
        StructField('col_b_num', DoubleType(), True), 
        StructField('col_c_num', DoubleType(), True), 
        StructField('partition_d_str', StringType(), False), 
        StructField('partition_e_str', StringType(), True)]) 

# Convert RDD to Spark DataFrame 
sdf = sqlContext.createDataFrame(my_rdd, schema=my_schema)

由于partition_e_str被宣布为nullable=True（用于StructField第三个参数），它写的时候有问题：当我转换RDD到火花数据帧，我提供的模式是这样产生的Hive表，因为它被用作分区字段之一。我把它改为：

# Define schema 
my_schema = StructType(
        [StructField('col_a_str', StringType(), False), 
        StructField('col_b_num', DoubleType(), True), 
        StructField('col_c_num', DoubleType(), True), 
        StructField('partition_d_str', StringType(), False), 
        StructField('partition_e_str', StringType(), False)])

，一切都很好再次！

课程：确保您的分区字段不为空！

来源

2017-05-19 02:01:52 vk1011

内存分配问题蜂巢表

回答

相关问题