问题标签 [foundry-python-transform]

我在我的 Foundry 实例中有一个使用增量计算的管道设置，但由于某种原因没有达到我的预期。即，我想读取我的转换的先前输出并获取日期的最大值，然后仅在该最大日期之后立即读取数据的输入。

出于某种原因，它没有达到我的预期，并且在构建/分析/修改代码过程中逐步执行代码非常令人沮丧。

我的代码如下所示：

我有一个创建两个数据集的数据连接源：

• 数据集 X（快照）
• 数据集 Y（增量）

这两个数据集来自同一来源。数据集X包含源表中所有行的当前状态。数据集Y提取自上次构建以来已更新的所有行。然后将这两个数据集在下游合并到 datasetZ中， datasetZ是 datasetX或 dataset 中每一行的最新版本Y。这使我们既可以进行低延迟更新，又可以保持良好的分区。

在源表中删除行时，这些行不再存在于 dataset 中，X但仍存在于 dataset 中Y。

将这些“已删除”行保留在数据集中的最佳方法是什么Z？理想情况下，我还可以在Y不丢失任何“已删除”行的情况下对数据集进行快照。

我想采用任意一组模式并将它们组合成一个可以在以后取消透视的数据集。最稳定的方法是什么？

假设我有数十个具有不同模式的输入。这些都有不同的列，实际上意思相同，只是名称不同。

修复列名后，我想创建作为这些列集合的输出，即动态旋转。我希望我的输入组合在一起后有一个单一的模式，我可以在以后进行动态旋转。

我真的不想经历并复制输入或创建数百个中间数据集，那么我最好如何做到这一点？

我想对我的数据集进行 Hive 分区，但我不太清楚如何确保拆分中的文件数是合理的。我知道我应该大致针对大小为 128MB 的文件

如何安全地缩放和控制 Hive 分区数据集文件内的行数？

我想测试 PySpark 正则表达式的不同输入，以查看它们在运行构建之前是否失败/成功。在运行完整的构建/检查之前，有没有办法在 Foundry 中对此进行测试？

我想使用 解析一系列.csv文件spark.read.csv，但我想在文件中包含每一行的行号。

我知道 Spark 通常不会订购 DataFrame，除非明确告知这样做，而且我不想编写自己的.csv文件解析器，因为这将比 Spark 自己的实现慢得多。如何以分布式安全方式添加此行号？

通过阅读zipWithIndex，它似乎很有用，但不幸的是，它似乎需要分区顺序稳定

我有一个使用 、 和 运行的构建DRIVER_MEMORY_LARGE，NUM_EXECUTORS_64为什么EXECUTOR_CORES_LARGE这没有足够的资源来防止我的工作由于OOM导致的 executor 丢失而失败？

我知道我的工作中有一些大任务，但这些资源不足以成功似乎很奇怪。

我有一组.xml要解析的文档。

我以前曾尝试使用获取文件内容并将它们转储到单个单元格中的方法来解析它们，但是我注意到这在实践中不起作用，因为我看到运行时间越来越慢，通常需要完成一项任务运行数十小时：

我的第一个转换获取.xml内容并将其放入单个单元格中，第二个转换获取此字符串并使用 Python 的xml库将字符串解析为文档。然后我可以从这个文档中提取属性并返回一个 DataFrame。

我正在使用UDF来执行将字符串内容映射到我想要的字段的过程。

我怎样才能使它更快/更好地处理大.xml文件？

我的 Foundry 转换在不同的运行中产生不同数量的数据，但我希望每个文件中的行数相似。我可以使用DataFrame.count()然后合并/重新分区，但这需要计算完整的数据集，然后缓存或重新计算。Spark有办法解决这个问题吗？

我想df.count()在我的 DataFrame 上运行，但我知道我的数据集总大小非常大。这是否存在将数据具体化回驱动程序的风险/增加驱动程序OOM的风险？