问题标签 [foundry-python-transform]

我有一个数据集，我想将每个列的唯一值均匀地重新分区为 10 个桶，并且我想将此结果调整为大量分区，以便每个分区都很小。

col_1保证是中的值之一["CREATE", "UPDATE", "DELETE"]

我的代码如下所示：

我看到我的大多数任务都以零行数据运行并完成，我希望数据均匀分布在我的partition_col1000 个分区中吗？

我从一个大的 csvs zip 文件开始，我在 Palantir Foundry 中解压缩了它。

我现在有一个由多个 csv 组成的数据集（每年一个），其中 csv 几乎是相同的模式，但有一些差异。如何将架构单独应用于每个 csv 或规范化它们之间的架构？

我已经阅读了 Foundry 中的文档以了解两者之间的区别，但我想知道在什么情况下我希望将STATIC_ALLOCATION配置文件应用于我的构建以防止我的执行程序被抢占。

在动态分配模式下运行时，我还有什么其他需要注意的吗？

我注意到我的代码仓库警告我在 for/while 循环中使用 withColumn 是一种反模式。为什么不建议这样做？这不是 PySpark API 的正常使用吗？

我有一些 PySpark 代码正在编写，我想在其中执行连接和其他操作，但我想在此阶段成功完成时记录下来。

为什么我没有看到按我期望的顺序记录？即使我的工作仍在继续工作，似乎一切都立即出现了……

我正在使用各种 Magritte 导出任务将 Foundry 数据集中的数据以parquet格式导出到 ABFS 系统（但 SFTP、S3、HDFS 和其他基于文件的导出也会出现同样的问题）。

我要导出的数据集相对较小，小于 512 MB，这意味着它们实际上不需要拆分到多个 parquet 文件中，将所有数据放在一个文件中就足够了。我通过以 a 结束之前的转换来完成此操作，以.coalesce(1)将所有数据放在一个文件中。

问题是：

• 默认情况下，文件名是part-0000-<rid>.snappy.parquet，每个版本都有不同的删除。这意味着，无论何时上传新文件，它都会与其他文件出现在同一个文件夹中，判断哪个是最新版本的唯一方法是最后修改日期。
• 每个版本的数据都存储在我的外部系统中，这会占用不必要的存储空间，除非我经常进入并删除旧文件。

所有这些都是不必要的复杂性被添加到我的下游系统中，我只想能够在一个步骤中提取最新版本的数据。

我在我的存储库中看到它警告我要使用union，而我应该使用unionByName. 这些不是一回事吗？我为什么要关心使用哪一个？

我很难理解为什么我无法运行转换，在等待了这么多分钟（有时是几个小时）后，它返回错误“序列化结果太大”。

在转换中，我有一个日期列表，我在 for 循环中迭代以在特定时间间隔内进行增量计算。

预期的数据集是迭代数据集的联合，应该包含 450k 行，不是太多，但我有很多计算阶段、任务和尝试！

配置文件已设置为中等配置文件，我无法在其他配置文件上进行缩放，也无法设置 maxResultSize = 0。

代码示例：

不要注意语法。这只是一个例子，说明循环内部有一系列操作。我想要的输出是一个数据框，其中包含每次迭代的数据框！

谢谢！！

当我的 spark 作业失败时，我收到以下错误**"org.apache.spark.shuffle.FetchFailedException: The relative remote executor(Id: 21), which maintains the block data to fetch is dead."**

我的 Spark 工作概览

输入大小约为 35 GB

我已经广播将所有较小的表与母表连接成 a dataframe1，然后我将每个大表加盐，然后再dataframe1加入dataframe1（左表）。

使用的配置文件：

使用上述方法和配置文件，我能够将运行时间降低 50%，但由于 Executor Loss问题，我仍然遇到 Shuffle Stage Failing。

有什么办法可以解决这个问题吗？

我调查了我的工作，并确定我确实有一个倾斜的任务。如何确定导致偏斜的此任务中的实际值是什么？

我的 Python 转换代码如下所示：