mongo-spark 和 spark-redshift 连接器的 spark sql 连接性能问题答案

【问题标题】：spark sql join performance issue with mongo-spark and spark-redshift connectorsmongo-spark 和 spark-redshift 连接器的 spark sql 连接性能问题
【发布时间】：2017-08-13 06:47:01
【问题描述】：

我们将 Apache-spark 与 mongo-spark library（用于连接 MongoDB）和 spark-redshift library（用于连接 Amazon Redshift DWH）一起使用。而且我们的工作表现非常糟糕。

所以我希望得到一些帮助，以了解我们是否在程序中做错了什么，或者这是我们所使用的基础设施所期望的。

我们在 4 个 AWS EC2 节点上使用 MESOS 资源管理器运行我们的工作，每个节点的配置如下：

RAM: 16GB, CPU cores: 4, SSD: 200GB

我们在 Redshift 集群中有 3 个表：

TABLE_NAME  SCHEMA                                    NUMBER_OF_ROWS
table1      (table1Id, table2FkId, table3FkId, ...)    50M
table2      (table2Id, phonenumber, email,...)         700M
table3      (table3Id, ...)                            2K

在 MongoDB 中，我们有一个包含 3500 万个文档的集合，其中一个示例文档如下（所有电子邮件和电话号码在这里都是唯一的，没有重复）：

{
  "_id": "19ac0487-a75f-49d9-928e-c300e0ac7c7c",
  "idKeys": {
    "email": [
      "a@gmail.com",
      "b@gmail.com"
    ],
    "phonenumber": [
      "1111111111",
      "2222222222"
    ]
  },
  "flag": false,
  ...
  ...
  ...
}

我们使用 spark-mongo 连接器过滤和展平（见 mongo-spark 聚合管道末尾的代码）到以下格式（因为我们需要将来自 Redshift 和 Mongo ON 电子邮件或电话号码的数据与另一个匹配可用的选项是 spark SQL 中的 array_contains()，它有点慢）：

  {"_id": "19ac0487-a75f-49d9-928e-c300e0ac7c7c", "email": "a@gmail.com", "phonenumber": null},
  {"_id": "19ac0487-a75f-49d9-928e-c300e0ac7c7c","email": "b@gmail.com","phonenumber": null},
  {"_id": "19ac0487-a75f-49d9-928e-c300e0ac7c7c","email": null,"phonenumber": "1111111111"},
  {"_id": "19ac0487-a75f-49d9-928e-c300e0ac7c7c","email": null,"phonenumber": "22222222222"}

Spark 计算步骤（请参阅下面的代码以更好地理解这些步骤）：

首先，我们使用 spark-redshift 连接器将 3 个 Redshift 表中的所有数据分别加载到 table1Dataset、table2Dataset、table3Dataset 中。
使用 SparkSQL 连接这 3 个表并创建新的数据集 redshiftJoinedDataset。（此操作在 6 小时内独立完成）
使用 mongo-spark 连接器将 MongoDB 数据加载到 mongoDataset。
加入 mongoDataset 和 redshiftJoinedDataset。（这是瓶颈，因为我们需要将来自 redshift 的超过 5000 万行与来自 mongodb 的超过 1 亿平展行连接起来）
注意：- mongo-spark 似乎也有一些 internal issue with its aggregation pipeline execution 这可能会使其非常慢。
然后我们正在对 finalId 上的数据进行一些聚合和分组

这是上述步骤的代码：

import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SparkSession;
import org.bson.Document;

import java.util.Arrays;

public class SparkMongoRedshiftTest {

    private static SparkSession sparkSession;
    private static SparkContext sparkContext;
    private static SQLContext sqlContext;

    public static void main(String[] args) {

        sparkSession = SparkSession.builder().appName("redshift-spark-test").getOrCreate();
        sparkContext = sparkSession.sparkContext();
        sqlContext = new SQLContext(sparkContext);


        Dataset table1Dataset = executeRedshiftQuery("(SELECT table1Id,table2FkId,table3FkId FROM table1)");
        table1Dataset.createOrReplaceTempView("table1Dataset");

        Dataset table2Dataset = executeRedshiftQuery("(SELECT table2Id,phonenumber,email FROM table2)");
        table2Dataset.createOrReplaceTempView("table2Dataset");

        Dataset table3Dataset = executeRedshiftQuery("(SELECT table3Id FROM table3");
        table3Dataset.createOrReplaceTempView("table3Dataset");


        Dataset redshiftJoinedDataset = sqlContext.sql(" SELECT a.*,b.*,c.*" +
                                                       " FROM table1Dataset a " +
                                                       " LEFT JOIN table2Dataset b ON a.table2FkId = b.table2Id" +
                                                       " LEFT JOIN table3Dataset c ON a.table3FkId = c.table3Id");
        redshiftJoinedDataset.createOrReplaceTempView("redshiftJoinedDataset");

        JavaMongoRDD<Document> userIdentityRDD = MongoSpark.load(getJavaSparkContext());
        Dataset mongoDataset = userIdentityRDD.withPipeline(
                Arrays.asList(
                        Document.parse("{$match: {flag: false}}"),
                        Document.parse("{ $unwind: { path: \"$idKeys.email\"  } }"),
                        Document.parse("{$group: {_id: \"$_id\",emailArr: {$push: {email: \"$idKeys.email\",phonenumber: {$ifNull: [\"$description\", null]}}},\"idKeys\": {$first: \"$idKeys\"}}}"),
                        Document.parse("{$unwind: \"$idKeys.phonenumber\"}"),
                        Document.parse("{$group: {_id: \"$_id\",phoneArr: {$push: {phonenumber: \"$idKeys.phonenumber\",email: {$ifNull: [\"$description\", null]}}},\"emailArr\": {$first: \"$emailArr\"}}}"),
                        Document.parse("{$project: {_id: 1,value: {$setUnion: [\"$emailArr\", \"$phoneArr\"]}}}"),
                        Document.parse("{$unwind: \"$value\"}"),
                        Document.parse("{$project: {email: \"$value.email\",phonenumber: \"$value.phonenumber\"}}")
                )).toDF();
        mongoDataset.createOrReplaceTempView("mongoDataset");

        Dataset joinRedshiftAndMongoDataset = sqlContext.sql(" SELECT a.* , b._id AS finalId " +
                                                             " FROM redshiftJoinedData AS a INNER JOIN mongoDataset AS b " +
                                                             " ON b.email = a.email OR b.phonenumber = a.phonenumber");

        //aggregating joinRedshiftAndMongoDataset
        //then storing to mysql
    }

    private static Dataset executeRedshiftQuery(String query) {
        return sqlContext.read()
                .format("com.databricks.spark.redshift")
                .option("url", "jdbc://...")
                .option("query", query)
                .option("aws_iam_role", "...")
                .option("tempdir", "s3a://...")
                .load();
    }

    public static JavaSparkContext getJavaSparkContext() {
        sparkContext.conf().set("spark.mongodb.input.uri", "");
        sparkContext.conf().set("spark.sql.crossJoin.enabled", "true");
        return new JavaSparkContext(sparkContext);
    }
}

在上述基础设施上完成这项工作的时间估计超过 2 个月。

所以定量地总结连接：

RedshiftDataWithMongoDataJoin => (RedshiftDataJoin)                INNER_JOIN (MongoData)
                              => (50M LEFT_JOIN 700M LEFT_JOIN 2K) INNER_JOIN (~100M)
                              => (50M)                             INNER_JOIN (~100M)

我们将不胜感激。

【问题讨论】：

第 3 阶段从 MongoDB 加载数据需要多长时间？分区器根本不应该是瓶颈，可能不是，但你的数字是多少？您是否检查过是通过聚合框架在 MongoDB 中进行数据转换更快还是在 Spark 中本地进行？
嗨@Ross 我没有在整个数据上单独测试过 mongo 聚合管道，但我已经在来自 Mongo 控制台和通过 spark-mongo 连接器的有限数据（仅在 1000 个文档上）进行了测试。加载似乎更快，但加入和内部改组可能会使其变慢。这项工作正在创建大约 16000 个分区。
@Ross 正如您所说“分区器根本不应该成为瓶颈，并且可能不是”，但是正如您在我的问题中看到的那样，我能够执行 LEFT_JOIN (50M X 700M X 2K) 在 6 小时内从 Redshift 加载数据，然后将这 50M 行与 100M mongo 文档连接起来。由于 mongo-spark 聚合，瓶颈可能没有发生，但我无法在其他任何地方找到问题。非常感谢您对这个问题的帮助。
当我在spark-mongo aggregation pipeline issue 上阅读您的 cmets 时，您说聚合管道被忽略了，所以在这种情况下，整个数据是否被加载到内存中，然后管道执行？还是管道被完全忽略？
分区器忽略当前版本中的管道，这意味着您可能会得到空分区，这些处理成本会很小，但在下一个版本中不会发生。但是，您只会将需要执行的数据加载到 Spark 中。数据的洗牌可能是一个问题 DAG 是什么样的？ databricks.com/blog/2015/06/22/…

标签： java mongodb apache-spark apache-spark-sql amazon-redshift

【解决方案1】：

因此，经过大量调查，我们了解到 table2 中 90% 的数据的电子邮件或电话号码为空，而我错过了处理查询中空值的连接。

所以这是这个性能瓶颈的主要问题。

修复此问题后，作业现在可在 2 小时内运行。

所以 spark-redshift 或 mongo-spark 没有问题，它们的表现非常好:)

【讨论】：