spark性能问题-将分区作为单个文件写入s3

jtoj6r0c  于 2021-05-27  发布在  Spark
关注(0)|答案(2)|浏览(518)

我正在运行一个spark作业,它的任务是扫描一个大文件并将其拆分为更小的文件。这个文件是json行格式的,我试图用一个特定的列(id)对它进行分区,并将每个分区作为一个单独的文件保存到s3中。文件大小约为12GB,但id的不同值约为500000个。查询大约需要15个小时。我能做些什么来提高性能?对于这样的任务,spark是一个糟糕的选择吗?请注意,我有权确保源代码为每个id的固定行数。

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from pyspark.sql import functions as F
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.sql.window import Window
from awsglue.utils import getResolvedOptions
from awsglue.transforms import *
from pyspark.sql.functions import udf, substring, instr, locate
from datetime import datetime, timedelta

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

# Get parameters that were passed to the job

args = getResolvedOptions(sys.argv, ['INPUT_FOLDER', 'OUTPUT_FOLDER', 'ID_TYPE', 'DATASET_DATE'])

id_type = args["ID_TYPE"]
output_folder = "{}/{}/{}".format(args["OUTPUT_FOLDER"], id_type, args["DATASET_DATE"])
input_folder = "{}/{}/{}".format(args["INPUT_FOLDER"], id_type, args["DATASET_DATE"])

INS_SCHEMA = StructType([
    StructField("camera_capture_timestamp", StringType(), True),
    StructField(id_type, StringType(), True),
    StructField("image_uri", StringType(), True)
])

data = spark.read.format("json").load(input_folder, schema=INS_SCHEMA)

data = data.withColumn("fnsku_1", F.col("fnsku"))

data.coalesce(1).write.partitionBy(["fnsku_1"]).mode('append').json(output_folder)

我也尝试过重新划分而不是合并。
我在用aws胶水

qhhrdooz

qhhrdooz1#

如果你不打算使用spark来做任何事情,而只是将文件分割成更小的版本,那么我会说spark是一个糟糕的选择。您最好在aws中按照下面的堆栈溢出文章中给出的方法来做这件事
假设您有一个可用的ec2示例,您可以运行如下操作:

aws s3 cp s3://input_folder/12GB.json - | split -l 1000 - output.
aws s3 cp output.* s3://output_folder/

如果您希望在spark中对数据进行进一步的处理,那么您需要将数据重新划分为128mb到1gb之间的块。使用默认(snappy)压缩,通常会得到原始文件大小的20%。所以,在您的例子中:在(12/5)~3和(12/5/8)~20个分区之间,所以:

data = spark.read.format("json").load(input_folder, schema=INS_SCHEMA) 

dataPart = data.repartition(12)

对于spark来说,这实际上不是一个特别大的数据集,处理起来也不应该那么麻烦。
保存为Parquet给你一个很好的恢复点,重新读取数据将非常快。总文件大小约为2.5 gb。

aiazj4mn

aiazj4mn2#

请考虑以下选项之一。如果能帮上忙那就太棒了:)
首先,如果合并,如注解中所说的@lamanus,这意味着您将减少分区的数量,因此也将减少writer任务的数量,从而将所有数据洗牌为1个任务。这可能是改善的第一个因素。
为了克服这个问题,即每个分区写一个文件并保持并行化级别,您可以更改以下逻辑:

object TestSoAnswer extends App {

  private val testSparkSession = SparkSession.builder()
    .appName("Demo groupBy and partitionBy").master("local[*]")
    .getOrCreate()
  import testSparkSession.implicits._

  // Input dataset with 5 partitions
  val dataset = testSparkSession.sparkContext.parallelize(Seq(
    TestData("a", 0), TestData("a", 1), TestData("b", 0), TestData("b", 1),
    TestData("c", 1), TestData("c", 2)
  ), 5).toDF("letter", "number")

  dataset.as[TestData].groupByKey(row => row.letter)
    .flatMapGroups {
      case (_, values) => values
    }.write.partitionBy("letter").mode("append").json("/tmp/test-parallel-write")

}

case class TestData(letter: String, number: Int)

它是如何工作的?

首先,代码执行无序排列,将与特定键(与分区相同)相关的所有行收集到相同的分区。因此,它将一次对属于该键的所有行执行写操作。前段时间我写了一篇关于 partitionBy 方法。大致来说,它会在内部对给定分区上的记录进行排序,然后将它们逐个写入文件中。
这样我们就得到了这样一个计划,其中只有1次洗牌,因此存在处理消耗操作:

== Physical Plan ==

* (2) SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, knownnotnull(assertnotnull(input[0, TestData, true])).letter, true, false) AS letter#22, knownnotnull(assertnotnull(input[0, TestData, true])).number AS number#23]

+- MapGroups TestSoAnswer$$$Lambda$1236/295519299@55c50f52, value#18.toString, newInstance(class TestData), [value#18], [letter#3, number#4], obj#21: TestData
   +- *(1) Sort [value#18 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(value#18, 200), true, [id=#15]
         +- AppendColumnsWithObject TestSoAnswer$$$Lambda$1234/1747367695@6df11e91, [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, knownnotnull(assertnotnull(input[0, TestData, true])).letter, true, false) AS letter#3, knownnotnull(assertnotnull(input[0, TestData, true])).number AS number#4], [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true, false) AS value#18]
            +- Scan[obj#2]

输出 TestSoAnswer 执行两次看起来是这样的:

test-parallel-write % ls
_SUCCESS letter=a letter=b letter=c
test-parallel-write % ls letter=a
part-00170-68245d8b-b155-40ca-9b5c-d9fb746ac76c.c000.json part-00170-cd90d64f-43c6-4582-aae6-fe443b6617f4.c000.json

test-parallel-write % ls letter=b
part-00161-68245d8b-b155-40ca-9b5c-d9fb746ac76c.c000.json part-00161-cd90d64f-43c6-4582-aae6-fe443b6617f4.c000.json

test-parallel-write % ls letter=c
part-00122-68245d8b-b155-40ca-9b5c-d9fb746ac76c.c000.json part-00122-cd90d64f-43c6-4582-aae6-fe443b6617f4.c000.json

您还可以使用此配置控制每个文件写入的记录数。
编辑:没有看到@mazaneicha的评论,但实际上,你可以试试 repartition("partitioning column") ! 这比分组表达式更清楚。
最好的,
巴托斯。

相关问题