将两个Spark mllib管道连接在一起

Question

我有两个独立的 DataFrames，每个都有几个不同的处理阶段，我使用 mllib 转换器在管道中进行处理。

现在我想要将这两个管道合并在一起，并保留每个 DataFrame 的特征（列）。

Scikit-learn 有 FeatureUnion 类来处理这个问题，但我似乎找不到 mllib 中相应的类似功能。

我可以在其中一个管道的末尾添加一个自定义转换器阶段，将另一个管道生成的 DataFrame 作为属性传入，并在 transform 方法中加以合并，但这看起来很混乱。

- Anake

---

你是在寻找连接(join)还是合并(union)？这两种操作都可以使用数据框(dataframes)来处理。 - jamborta

@jamborta 这是一个连接操作，但是我想将其作为管道阶段来执行，以便我可以在整个管道中进行模式检查。 - Anake

Answer 1

Pipeline或PipelineModel是有效的PipelineStages，因此可以组合在一个单一的Pipeline中。例如：Pipeline可以如下所示：

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler

df = spark.createDataFrame([
    (1.0, 0, 1, 1, 0),
    (0.0, 1, 0, 0, 1)
], ("label", "x1", "x2", "x3", "x4"))

pipeline1 = Pipeline(stages=[
    VectorAssembler(inputCols=["x1", "x2"], outputCol="features1")
])

pipeline2 = Pipeline(stages=[
    VectorAssembler(inputCols=["x3", "x4"], outputCol="features2")
])

您可以组合管道：

Pipeline(stages=[
    pipeline1, pipeline2, 
    VectorAssembler(inputCols=["features1", "features2"], outputCol="features")
]).fit(df).transform(df)

+-----+---+---+---+---+---------+---------+-----------------+
|label|x1 |x2 |x3 |x4 |features1|features2|features         |
+-----+---+---+---+---+---------+---------+-----------------+
|1.0  |0  |1  |1  |0  |[0.0,1.0]|[1.0,0.0]|[0.0,1.0,1.0,0.0]|
|0.0  |1  |0  |0  |1  |[1.0,0.0]|[0.0,1.0]|[1.0,0.0,0.0,1.0]|
+-----+---+---+---+---+---------+---------+-----------------+

或预安装的PipelineModels：

model1 = pipeline1.fit(df)
model2 = pipeline2.fit(df)

Pipeline(stages=[
    model1, model2, 
    VectorAssembler(inputCols=["features1", "features2"], outputCol="features")
]).fit(df).transform(df)

+-----+---+---+---+---+---------+---------+-----------------+
|label| x1| x2| x3| x4|features1|features2|         features|
+-----+---+---+---+---+---------+---------+-----------------+
|  1.0|  0|  1|  1|  0|[0.0,1.0]|[1.0,0.0]|[0.0,1.0,1.0,0.0]|
|  0.0|  1|  0|  0|  1|[1.0,0.0]|[0.0,1.0]|[1.0,0.0,0.0,1.0]|
+-----+---+---+---+---+---------+---------+-----------------+

因此，我建议的方法是预先加入数据，并对整个DataFrame进行fit和transform操作。

另请参见:

• Apack Spark 如何在无需再次拟合的情况下向现有PipelineModel添加新的fitted stage

- zero323

---

这里我们有一个单独的数据框，但是假设我们有两个不同行数的不同数据框，Spark ML Pipeline 是如何在内部连接它们的呢？它是交叉连接还是其他方式？ - Kaushal

使用两个独立的管道和使用一个单一的管道，是否存在性能差异？ - Tim