草庐IT

关于scala：如何优化Spark以将大量数据写入S3

codeneng 2023-03-28 原文

How to optimize Spark for writing large amounts of data to S3

我在 EMR 上使用 Apache Spark 进行了大量 ETL。

我对获得良好性能所需的大部分调整都相当满意，但我有一项工作似乎无法弄清楚。

基本上，我使用了大约 1 TB 的 parquet 数据 - 分布在 S3 中的数万个文件中 - 并添加了几列并将其写出，并按数据的日期属性之一进行分区 - 再次，parquet 格式在 S3 中。

我是这样跑的：

1
spark-submit --conf spark.dynamicAllocation.enabled=true  --num-executors 1149 --conf spark.driver.memoryOverhead=5120 --conf  spark.executor.memoryOverhead=5120 --conf  spark.driver.maxResultSize=2g --conf spark.sql.shuffle.partitions=1600 --conf spark.default.parallelism=1600 --executor-memory 19G --driver-memory 19G --executor-cores 3 --driver-cores 3 --class com.my.class path.to.jar <program args>

集群的大小是根据输入数据集的大小动态确定的，num-executors、spark.sql.shuffle.partitions和spark.default.parallelism参数是根据输入数据集的大小来计算的集群。

代码大致是这样的：

1
2
3
4
5
6
7
8
9
va df = (read from s3 and add a few columns like timestamp and source file name)

val dfPartitioned = df.coalesce(numPartitions)

val sqlDFProdDedup = spark.sql(s""" (query to dedup against prod data""");

sqlDFProdDedup.repartition($"partition_column")
  .write.partitionBy("partition_column")
  .mode(SaveMode.Append).parquet(outputPath)

当我查看神经节图时，我得到一个巨大的资源峰值，而重复数据删除逻辑运行和一些数据洗牌，但数据的实际写入只使用了一小部分资源并运行了几个小时.

我不认为主要问题是分区倾斜，因为数据应该公平地分布在所有分区中。

分区列本质上是一个月中的一天，因此每个作业通常只有 5-20 个分区，具体取决于输入数据集的跨度。每个分区通常在 10-20 个 parquet 文件中包含大约 100 GB 的数据。

我正在设置 spark.sql.files.maxRecordsPerFile 来管理这些输出文件的大小。

所以，我最大的问题是：如何提高这里的性能？

简单地添加资源似乎没有多大帮助。

我已经尝试让执行器更大(以减少洗牌)并增加每个执行器的 CPU 数量，但这似乎并不重要。

提前致谢！

  • 一个后续说明：有问题的表有大约 80 行。其中 2 行是大字符串：一行大约 100 个字符，另一行大约 1000 个字符。根据经验，一旦采用(活泼的)镶木地板格式，如果没有这两列，数据大date小 1/3。不管那值多少钱...

Zack，我有一个类似的用例，每天要处理的文件要多"n"倍。我将假设您按原样使用上面的代码并尝试提高整体工作的性能。以下是我的一些观察：

  • 不确定 coalesce(numPartitions) 数字实际上是什么以及为什么在重复数据删除过程之前使用它。您的 spark-submit 显示您正在创建 1600 个分区，这已经足够开始了。

  • 如果您要在写入之前重新分区，那么上面的合并可能根本没有好处，因为重新分区会打乱数据。

  • 由于您声称要编写 10-20 个 parquet 文件，这意味着您在工作的最后一部分中仅使用 10-20 个内核进行编写，这是其缓慢的主要原因。根据 100 GB 的估计，镶木地板文件的范围从大约 5GB 到 10 GB，这真的很大，我怀疑除非他们使用 EMR 或类似的设备(如果读取，执行程序内存很大整个文件或溢出到磁盘)，因为内存要求太高。我建议创建大约 1GB 的镶木地板文件以避免任何这些问题。

    • 此外，如果您创建 1GB parquet 文件，您可能会将该过程加快 5 到 10 倍，因为您将使用更多的执行器/内核来并行编写它们。您实际上可以通过简单地使用默认分区编写数据帧来运行实验。

    这让我明白了你真的不需要使用重新分区，因为你想 write.partitionBy("partition_date") 调用。您的 repartition() 调用实际上是强制数据帧最多只有 30-31 个分区，具体取决于该月的天数，这是驱动正在写入的文件数量的原因。 write.partitionBy("partition_date") 实际上是在 S3 分区中写入数据，如果您的数据帧有 90 个分区，它将写入速度快 3 倍(3 * 30)。 df.repartition() 迫使它放慢速度。您真的需要 5GB 或更大的文件吗？

  • 另一个重点是 Spark 惰性求值有时过于聪明。在您的情况下，它很可能只使用基于 repartition(number) 的整个程序的执行程序数量。相反，您应该尝试 df.cache() -> df.count() and then df.write()。它的作用是强制 spark 使用所有可用的执行器核心。我假设您正在并行读取文件。在您当前的实现中，您可能使用 20-30 个内核。需要注意的是，当您使用 r4/r5 机器时，请随意将您的执行程序内存增加到 8 核的 48G。我发现 8 核比标准的 5 核推荐更快地完成我的任务。

  • 另一个指针是尝试 ParallelGC 而不是 G1GC。对于像这样的用例，当您读取 1000 倍文件时，我注意到它的性能比 G1Gc 更好或不差。请试一试。

    • 在我的工作负载中，我使用基于 coalesce(n) 的方法，其中 \\'n\\' 给了我一个 1GB 的 parquet 文件。我使用集群上所有可用的内核并行读取文件。仅在写入部分，我的内核处于空闲状态，但您无法避免这种情况。

    我不确定 spark.sql.files.maxRecordsPerFile 如何与 coalesce() or repartition() 结合使用，但我发现 1GB 似乎可以用于 pandas、Redshift 频谱、Athena 等。

    希望对您有所帮助。
    查鲁

    这里有一些优化以加快运行速度。

    (1) 文件提交者 - 这是 Spark 将部分文件读取到 S3 存储桶的方式。每个操作都是不同的，并且将基于

    1
    spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version 2

    说明

    这会将文件直接写入零件文件，或者最初将它们加载到临时文件并将它们复制到它们的最终状态零件文件中。

    (2) 对于文件大小，您可以根据每条记录的平均字节数得出它。下面我计算每条记录的字节数以计算 1024 MB 的记录数。我会先尝试每个分区 1024MB，然后向上移动。

    1
    2
    3
    4
    5
    6
    import org.apache.spark.util.SizeEstimator

    val numberBytes : Long = SizeEstimator.estimate(inputDF.rdd)
    val reduceBytesTo1024MB = numberBytes/123217728
    val numberRecords = inputDF.count
    val recordsFor1024MB = (numberRecords/reduceBytesTo1024MB).toInt + 1

    (3) [我还没有尝试过] EMR Committer - 如果您使用的是 EMR 5.19 或更高版本，因为您正在输出 Parquet。您可以将 Parquet 优化编写器设置为 TRUE。

    1
    spark.sql.parquet.fs.optimized.committer.optimization-enabled true

    • 谢谢。我会试试这些！
    • 您可以尝试在写入之前对数据帧进行持久化和运行计数，以减少固有的 Spark I/O 操作。您还在使用 s3a、s3n 或 s3 策略进行写作吗？
    • 策略 (1) 仍然是重命名提交，因此 O(data) - 并且具有错误的失败语义；如果您使用的是 Hadoop-3.1 JAR，您可以切换到 S3A 提交程序(也需要调整 spark 的版本来支持它)
    • 从我在 google 上看到的情况来看，使用在 EMR 上本地运行的 Spark 包不支持 S3A - 我必须使用(并支付)来自 DataBricks 的优化 Spark。这听起来正确吗？
    • 仅作为更新：我尝试了此处建议的所有优化，但没有明显的改进

    scalaSparkspanclasssectionamazon-emramazon-s3apache-spark

    有关关于scala：如何优化Spark以将大量数据写入S3的更多相关文章

    1. ruby - 如何使用 Nokogiri 的 xpath 和 at_xpath 方法 - 2

      我正在学习如何使用Nokogiri，根据这段代码我遇到了一些问题:require'rubygems'require'mechanize'post_agent=WWW::Mechanize.newpost_page=post_agent.get('http://www.vbulletin.org/forum/showthread.php?t=230708')puts"\nabsolutepathwithtbodygivesnil"putspost_page.parser.xpath('/html/body/div/div/div/div/div/table/tbody/tr/td/div

    2. ruby - 如何从 ruby​​ 中的字符串运行任意对象方法？ - 2

      总的来说，我对ruby​​还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用

    3. python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声？ - 2

      关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题？更新问题，以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点？我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等)，但我确实想创建一个输出文件。

    4. ruby-on-rails - 如何验证 update_all 是否实际在 Rails 中更新 - 2

      给定这段代码defcreate@upgrades=User.update_all(["role=?","upgraded"],:id=>params[:upgrade])redirect_toadmin_upgrades_path,:notice=>"Successfullyupgradeduser."end我如何在该操作中实际验证它们是否已保存或未重定向到适当的页面和消息？ 最佳答案 在Rails3中，update_all不返回任何有意义的信息，除了已更新的记录数(这可能取决于您的DBMS是否返回该信息)。http://ar.ru

    5. ruby-on-rails - 'compass watch' 是如何工作的/它是如何与 rails 一起使用的 - 2

      我在我的项目目录中完成了compasscreate.和compassinitrails。几个问题:我已将我的.sass文件放在public/stylesheets中。这是放置它们的正确位置吗？当我运行compasswatch时，它不会自动编译这些.sass文件。我必须手动指定文件:compasswatchpublic/stylesheets/myfile.sass等。如何让它自动运行？文件ie.css、print.css和screen.css已放在stylesheets/compiled。如何在编译后不让它们重新出现的情况下删除它们？我自己编译的.sass文件编译成compiled/t

    6. ruby - 如何将脚本文件的末尾读取为数据文件(Perl 或任何其他语言) - 2

      我正在寻找执行以下操作的正确语法(在Perl、Shell或Ruby中):#variabletoaccessthedatalinesappendedasafileEND_OF_SCRIPT_MARKERrawdatastartshereanditcontinues. 最佳答案 Perl用__DATA__做这个:#!/usr/bin/perlusestrict;usewarnings;while(){print;}__DATA__Texttoprintgoeshere 关于ruby-如何将脚

    7. ruby - 如何指定 Rack 处理程序 - 2

      Rackup通过Rack的默认处理程序成功运行任何Rack应用程序。例如:classRackAppdefcall(environment)['200',{'Content-Type'=>'text/html'},["Helloworld"]]endendrunRackApp.new但是当最后一行更改为使用Rack的内置CGI处理程序时，rackup给出“NoMethodErrorat/undefinedmethod`call'fornil:NilClass”:Rack::Handler::CGI.runRackApp.newRack的其他内置处理程序也提出了同样的反对意见。例如Rack

    8. ruby - 如何每月在 Heroku 运行一次 Scheduler 插件？ - 2

      在选择我想要运行操作的频率时，唯一的选项是“每天”、“每小时”和“每10分钟”。谢谢!我想为我的Rails3.1应用程序运行调度程序。 最佳答案 这不是一个优雅的解决方案，但您可以安排它每天运行，并在实际开始工作之前检查日期是否为当月的第一天。 关于ruby-如何每月在Heroku运行一次Scheduler插件？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.com/questions/8692687/

    9. ruby-on-rails - 如何从 format.xml 中删除 <hash></hash> - 2

      我有一个对象has_many应呈现为xml的子对象。这不是问题。我的问题是我创建了一个Hash包含此数据，就像解析器需要它一样。但是rails自动将整个文件包含在.........我需要摆脱type="array"和我该如何处理？我没有在文档中找到任何内容。 最佳答案 我遇到了同样的问题；这是我的XML:我在用这个:entries.to_xml将散列数据转换为XML，但这会将条目的数据包装到中所以我修改了:entries.to_xml(root:"Contacts")但这仍然将转换后的XML包装在“联系人”中，将我的XML代码修改为

    10. ruby - 如何使用文字标量样式在 YAML 中转储字符串？ - 2

      我有一大串格式化数据(例如JSON)，我想使用Psychinruby​​同时保留格式转储到YAML。基本上，我希望JSON使用literalstyle出现在YAML中:---json:|{"page":1,"results":["item","another"],"total_pages":0}但是，当我使用YAML.dump时，它不使用文字样式。我得到这样的东西:---json:!"{\n\"page\":1,\n\"results\":[\n\"item\",\"another\"\n],\n\"total_pages\":0\n}\n"我如何告诉Psych以想要的样式转储标量？解

    随机推荐