在使用 Hive EMR 一段时间后，我正在迈出使用 Spark 的第一步。
我想读取以以下格式保存到 S3 的 Hive 表：s3://<bucket>/<rootpath>/date=<date>/fileNames
我可以使用答案in this question，但随后我失去了 dataRows 与date 之间的连接，这是因为我没有将日期保存在文件中。
有没有一种简单的方法来获得每一行数据的文件名？
【问题讨论】：
标签： java hadoop amazon-web-services amazon-s3 apache-spark
如果您从文件名中需要的只是听起来的日期，那么您不需要将文件名作为 spark 获取，并且如果您正确创建表，hive 会自动为您执行此操作。让我演示一下：
# in hive
hive> create table t1 ( name string) partitioned by (date string) STORED AS TEXTFILE LOCATION 'your s3 path';
# data.txt contains 'john' and 'jay' in two different lines
hive> load data local inpath 'data.txt' into table t1 PARTITION(date='2015-12-30');
hive> select * from t1;
OK
john 2015-12-30
jay 2015-12-30
# in spark-shell
scala> sqlContext.sql("select * from t1").foreach(println);
[john,2015-12-30]
[jay,2015-12-30]
我认为这是你想要的。它的另一个优点是您的数据可以在查询时利用分区的性能优势。
【讨论】：
您可以使用 wholeTextFiles 来读取 rdd。这将读取每个文件，文件名作为键，文件的全部内容作为值。从那里，您应该能够使用 flatMapValues 将每条记录分成自己的 k/v 对。
val input = sc.wholeTextFiles(s3://...)
val inputFlat = input.flatMapValues(line => line.split("\n"))
对于这个例子，如果你的路径是 /user/hive/date=December/part-0000 并且 part-0000 的内容是
Joe December-28 Something
Ryan December-29 AnotherThing
输出如下所示：
input.take(1)
(/user/hive/date=December/part-0000, Joe December-28 Something\n Ryan December-29 AnotherThing)
inputFlat.take(2)
(/user/hive/date=December/part-0000, Joe December-28 Something)
(/user/hive/date=December/part-0000, Ryan December-29 AnotherThing)
我想您可以尝试以下方法。读取记录会有点慢，但是重新分区后可以最大化并行处理
inputFlat.flatMapValues(//some split).repartition(numWorkers)
我们可以尝试的另一件事是使用它： 在 hive 中，您可以使用名为 INPUT__FILE__NAME 的虚拟列检索找到记录的文件，例如：
select INPUT__FILE__NAME, id, name from users where ...;
我不确定它是否会起作用，但您可以尝试在您的 .sql api 中使用它。您必须确保您的 sqlContext 具有 hive-site.xml。
【讨论】：