如何在pyspark中更改DataFrame的hdfs块大小

Question

这似乎与

有关

How to change hdfs block size in pyspark?

我可以用rdd.saveAsTextFile成功改变hdfs块大小， 但不是对应的DataFrame.write.parquet，无法以parquet格式保存。

不确定是pyspark DataFrame中的错误还是我没有正确设置配置。

以下是我的测试代码：

##########
# init
##########
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession

import hdfs
from hdfs import InsecureClient
import os

import numpy as np
import pandas as pd
import logging

os.environ['SPARK_HOME'] = '/opt/spark-2.2.1-bin-hadoop2.7'

block_size = 512 * 1024

conf = SparkConf().setAppName("myapp").setMaster("spark://spark1:7077").set('spark.cores.max', 20).set("spark.executor.cores", 10).set("spark.executor.memory", "10g").set("spark.hadoop.dfs.blocksize", str(block_size)).set("spark.hadoop.dfs.block.size", str(block_size))

spark = SparkSession.builder.config(conf=conf).getOrCreate()
spark.sparkContext._jsc.hadoopConfiguration().setInt("dfs.blocksize", block_size)
spark.sparkContext._jsc.hadoopConfiguration().setInt("dfs.block.size", block_size)

##########
# main
##########

# create DataFrame
df_txt = spark.createDataFrame([\{'temp': "hello"}, \{'temp': "world"}, \{'temp': "!"}])

# save using DataFrameWriter, resulting 128MB-block-size

df_txt.write.mode('overwrite').format('parquet').save('hdfs://spark1/tmp/temp_with_df')

# save using rdd, resulting 512k-block-size
client = InsecureClient('http://spark1:50070')
client.delete('/tmp/temp_with_rrd', recursive=True)
df_txt.rdd.saveAsTextFile('hdfs://spark1/tmp/temp_with_rrd')

Answer 1

从以下链接找到答案：

https://forums.databricks.com/questions/918/how-to-set-size-of-parquet-output-files.html

我可以使用 spark.hadoop.parquet.block.size 成功设置拼花块大小

以下是示例代码：

# init
block_size = 512 * 1024 

conf = SparkConf().setAppName("myapp").setMaster("spark://spark1:7077").set('spark.cores.max', 20).set("spark.executor.cores", 10).set("spark.executor.memory", "10g").set('spark.hadoop.parquet.block.size', str(block_size)).set("spark.hadoop.dfs.blocksize", str(block_size)).set("spark.hadoop.dfs.block.size", str(block_size)).set("spark.hadoop.dfs.namenode.fs-limits.min-block-size", str(131072))

sc = SparkContext(conf=conf) 
spark = SparkSession(sc) 

# create DataFrame 
df_txt = spark.createDataFrame([{'temp': "hello"}, {'temp': "world"}, {'temp': "!"}]) 

# save using DataFrameWriter, resulting 512k-block-size 

df_txt.write.mode('overwrite').format('parquet').save('hdfs://spark1/tmp/temp_with_df')

# save using DataFrameWriter.csv, resulting 512k-block-size 
df_txt.write.mode('overwrite').csv('hdfs://spark1/tmp/temp_with_df_csv') 

# save using DataFrameWriter.text, resulting 512k-block-size

df_txt.write.mode('overwrite').text('hdfs://spark1/tmp/temp_with_df_text')

# save using rdd, resulting 512k-block-size 
client = InsecureClient('http://spark1:50070') 
client.delete('/tmp/temp_with_rrd', recursive=True) 
df_txt.rdd.saveAsTextFile('hdfs://spark1/tmp/temp_with_rrd')

Answer 2

Hadoop 和 Spark 是两个独立的工具，它们有自己的工作策略。 Spark 和 Parquet 使用数据分区，块大小对它们没有意义。按照 Spark 所说的去做，然后在 HDFS 中做你想做的事情。

您可以通过以下方式更改 Parquet 分区号

df_txt.repartition(6).format("parquet").save("hdfs://...")