PySpark - 读取递归 Hive 表

Question

我有一个Hive表，它在HDFS中有多个子目录，如下所示：

/hdfs_dir/my_table_dir/my_table_sub_dir1
/hdfs_dir/my_table_dir/my_table_sub_dir2
...

通常，我在运行 Hive 脚本之前设置以下参数：

set hive.input.dir.recursive=true;
set hive.mapred.supports.subdirectories=true;
set hive.supports.subdirectories=true;
set mapred.input.dir.recursive=true;
select * from my_db.my_table;

我正在尝试使用 PySpark 做同样的事情，

conf = (SparkConf().setAppName("My App")
        ...
        .set("hive.input.dir.recursive", "true")
        .set("hive.mapred.supports.subdirectories", "true")
        .set("hive.supports.subdirectories", "true")
        .set("mapred.input.dir.recursive", "true"))
sc = SparkContext(conf = conf)
sqlContext = HiveContext(sc)
my_table = sqlContext.sql("select * from my_db.my_table")

并最终出现如下错误：

java.io.IOException: Not a file: hdfs://hdfs_dir/my_table_dir/my_table_sub_dir1

在

Spark 中读取带有子目录的 Hive 表的正确方法是什么？

Answer 1

我发现这些值前面必须有 spark，如下所示：

.set("spark.hive.mapred.supports.subdirectories","true")
.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive","true")

Answer 2

在执行查询之前，请尝试通过 ctx.sql（） 设置它们：

sqlContext.sql("SET hive.mapred.supports.subdirectories=true")
sqlContext.sql("SET mapreduce.input.fileinputformat.input.dir.recursive=true")
my_table = sqlContext.sql("select * from my_db.my_table")

Answer 3

尝试通过SpakSession设置它们以执行查询：

sparkSession = (SparkSession
                        .builder
                        .appName('USS - Unified Scheme of Sells')
                        .config("hive.metastore.uris", "thrift://probighhwm001:9083", conf=SparkConf())
                        .config("hive.input.dir.recursive", "true")
                        .config("hive.mapred.supports.subdirectories", "true")
                        .config("hive.supports.subdirectories", "true")
                        .config("mapred.input.dir.recursive", "true")
                        .enableHiveSupport()
                        .getOrCreate()
                        )