使用 sqlcontext 触发并行查询

Question

我们正在尝试在 EMR 中的 spark 上运行 ETL。 S3 中有大约 20 亿个事件作为 gzip 压缩的 json 行。它们总共大约有 30 个文件。我正在使用 pyspark。

这是代码，

def value_to_list(columns):
    def value_map(values):
        data = []
        for val in values:
            d = val.asDict()
            data.append([d[column] for column in columns])
        return data

    return value_map


def main():
    sc = SparkContext()
    sql_context = SQLContenxt(sc)

    all_events = SQLContenxt(sc).read.json("s3n://...", schema=StructType(fields), timestampFormat="yyyy-MM-dd HH:mm:ss")
    all_events.registerTempTable('allevents')

    for event_type in event_types:
        process_event(sc, event_type, "allevents")


def process_event(sparkcontext, etype, tablename):
    query = "select app_id, source, type, {time_cols}, count(*) as total " \
            "from {table} where  type = '{event_type}' " \
            "group by app_id, source, type, {time_cols}"
    time_cols_spec = [('hour', 'day', 'month', 'year'),
                      ('day', 'month', 'year'),
                      ('month', 'year'),
                      ('year')]

    for time_cols in time_cols_spec:
        final_query = query.format(time_cols=", ".join(time_cols),
                                   table=tablename,
                                   event_type=etype)
        dataframe = sql_context.sql(final_query)

        dataframe.rdd.groupBy(lambda r: r['app_id'])\
            .mapValues(value_to_list(['source'] + time_cols))\
            .saveAsTextFile("s3n://...")

所以，我们有大约 30 种类型的事件，并且对于每个事件，我将小时、日、月和年汇总为 4 种组合。所以每个查询4个。我们总共有大约 2000M 的事件。

我正在运行它

• AWS EMR (5.0.3)
• Apache Spark 2.0.1
• 1个master，2个worker
• 每台机器都是m3.2xlarge
• 总内存为 90GB

问题是，最终保存需要很长时间。上次我查询 2 次组合和 1 次事件需要 14 小时 :(

我知道我不会以并行方式进行。循环是顺序的。并且有2个循环。但我希望rdd,groupBy、mapValues 能够并行运行。当我看到事件时间线时，我看到它的 saveAsTextFile 占用了 99% 的时间。可能是因为 spark 执行得很慢。

我需要让这个过程并行且快速。我怎样才能做到这一点？

Answer 1

您可以应用 4 项主要优化：

1. 您正在对未针对查询进行优化的普通 json 文件执行聚合。将它们重写为 parquet，按事件类型重新分区并存储在 S3 上 - 它们将占用更少的空间，您的应用程序将获得不错的速度提升。

2. 增加并行度。无需在如此强大的 VM 上安装驱动程序（主），而是生成一个较小的实例（例如 m3.medium）并将所有 3 个大实例用于工作人员。

3. 将 RDD API 调用替换为 Dataframe：.rdd.groupBy().mapValues() 可以替换为 .groupBy(dataframe.app_id).agg(collect_list())，然后进行一些映射。

4. 您可以对（小时、天、月、年）数据集的原始数据执行查询，然后使用此聚合来计算给定事件的所有剩余查询。