python - Pyspark Invalid Input Exception try except 错误

Question

我正在尝试使用 pyspark 从 s3 读取最近 4 个月的数据并处理数据，但收到以下异常。

org.apache.hadoop.mapred.InvalidInputException: Input Pattern s3://path_to_clickstream/date=201508*

在每个月的第一天，由于 s3 路径中没有条目(一个单独的作业处理数据并将数据上传到 s3 路径，而我的作业在该路径之前运行)，作业失败。我想知道是否有办法捕获此异常并允许作业继续处理所有存在的路径？

Answer 1

您可以简单地尝试在加载后立即触发廉价操作并捕获 Py4JJavaError:

from py4j.protocol import Py4JJavaError

def try_load(path):
    rdd = sc.textFile(path)
    try:
        rdd.first()
        return rdd
    except Py4JJavaError as e:
        return sc.emptyRDD()

rdd = try_load(s3_path)
if not rdd.isEmpty():
    run_the_rest_of_your_code(rdd)

编辑:

如果你想处理多个路径，你可以分别处理每个路径并合并结果:

paths = [
    "s3://path_to_inputdir/month1*/",
    "s3://path_to_inputdir/month2*/",
    "s3://path_to_inpu‌​tdir/month3*/"]

rdds = sc.union([try_load(path) for path in paths])

如果你想要更好的控制，你可以list content并加载已知文件。

如果这些路径中至少有一个是非空的，你应该能够使事情变得更简单并像这样使用 glob:

sc.textFile("s3://path_to_inputdir/month[1-3]*/")