从文件创建输入变量的N个置换的PCollection最惯用的方法是什么？

Question

我正在尝试使用apachebeam来并行化模拟的N试验。这个模拟是在一个矩阵V上运行的，该矩阵来源于.matMATLAB文件。我的第一直觉（请原谅，我对apachebeam/Dataflow不熟悉）是扩展FileBasedSource，但进一步的调查使我确信事实并非如此。最明显的是，Apache Beam documentation表示，“如果您想使用源API提供的高级功能，您应该创建一个新的源代码”，而且我不需要它们中的任何一个—我只想读取一个变量（或几个变量）！我最终偶然发现了https://stackoverflow.com/a/45946643，这就是我现在打算如何得到V（将结果类似文件的对象从FileSystems.open传递到scipy.io.loadmat）。在

下一个问题是如何创建PCollection的N排列。显而易见的解决方案是beam.Create([permutation(V) for _ in xrange(N)])。然而，我对the documentation中的评论有点反感，“后者主要用于测试和调试目的。”也许稍微有一点改进是在DoFn中执行置换。在

我还有最后一个主意。这听起来有点愚蠢（很可能很愚蠢），但请听我说完这一点（幽默我睡眠不足的自我）。The documentation指向CountingSource（以及ReadFromCountingSource）的实现。使用ReadFromCountingSource(N)比使用beam.Create(range(N))有什么好处吗？如果是这样，以下（a的开头）Pipeline是否合理：

(p | 'trial_count' >> ReadFromCountingSource(N)
   | 'permute_v' >> beam.FlatMap(lambda _, x: permutation(x), V)
   | ...)

束流/数据流专家，你有什么建议？在

谢谢你！！在

Answer 1

让我重新表述一下我认为你在问什么，如果我错了，请纠正我：

在MATLAB文件中有一个矩阵V，需要读入该矩阵，然后运行模拟的N试验。在

编辑：FileBasedSource不能直接使用。我已经更正了下面的解释。

apachebeam有内置的PTransforms来读取许多文件格式，但没有MATLAB文件。您需要创建自己的源实现并读取转换。有很多基于FileBasedSource的例子，比如^{}和{a2}。在

Beam文档提供了实现新的I/O转换的技巧：https://beam.apache.org/documentation/io/authoring-overview/

如果与源代码解耦，则添加置换逻辑将更简单。如果试验次数N是静态的或在管道构建过程中已知的，则可以使用FlatMap和普通Python代码返回每个置换的iterable。所以你的逻辑看起来更像：

(p | 'read' >> ReadFromMatlab(file_name)
   | 'permute_v' >> beam.FlatMap(lambda x: permutation(x, N))
   | ...)

Answer 2

要将矩阵从.mat文件加载到PCollection，请从beam.Create派生一个PTransform包装器scipy.io.loadmat：

class LoadMat(beam.Create):

    def __init__(self, file_name, mdict=None, appendmat=True, **kwargs):
        mat_dict = scipy.io.loadmat(file_name, mdict, appendmat, **kwargs)
        super(LoadMat, self).__init__([mat_dict])

按如下方式调用此转换：

^{pr2}$