Elasticsearch + Apache Spark 性能

Question

我正在尝试使用 Apache spark 在 Elasticsearch 中查询我的数据，但我的 spark 作业需要大约 20 小时来进行聚合并且仍在运行。在 ES 中相同的查询大约需要 6 秒。

我了解数据必须从 Elasticsearch 集群移动到我的 spark 集群，并且在 Spark 中进行一些数据混洗。

我的 ES 索引中的数据约为。 3亿个文档，每个文档大约有400个字段（1.4Terabyte）。

我有一个 3 节点 spark 集群（1 个 master，2 个 worker），总共有 60GB 内存和 8 个内核。

运行时间不可接受，有没有办法让我的 spark 作业运行得更快？

这是我的火花配置：

SparkConf sparkConf = new SparkConf(true).setAppName("SparkQueryApp")
                 .setMaster("spark://10.0.0.203:7077")    
                 .set("es.nodes", "10.0.0.207")
                 .set("es.cluster", "wp-es-reporting-prod")              
                .setJars(JavaSparkContext.jarOfClass(Demo.class))
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .set("spark.default.parallelism", String.valueOf(cpus * 2))
                .set("spark.executor.memory", "8g");

已编辑

    SparkContext sparkCtx = new SparkContext(sparkConf);

    SQLContext sqlContext = new SQLContext(sparkCtx);
    DataFrame df = JavaEsSparkSQL.esDF(sqlContext, "customer-rpts01-201510/sample");

    DataFrame dfCleaned = cleanSchema(sqlContext, df);

    dfCleaned.registerTempTable("RPT");

    DataFrame sqlDFTest = sqlContext.sql("SELECT agent, count(request_type) FROM RPT group by agent");

    for (Row row : sqlDFTest.collect()) {
        System.out.println(">> " + row);
    }

Answer 1

我知道发生了什么，基本上，我试图操纵数据框架构，因为我有一些带点的字段，例如 user.firstname。 这似乎在火花的收集阶段引起了问题。为了解决这个问题，我只需要重新索引我的数据，这样我的字段就不再有点，而是下划线，例如 user_firstname。

Answer 2

恐怕您无法在仅 120 GB 总 RAM 的情况下执行超过 1.4 TB 的组并获得良好的性能。 DF 将尝试将所有数据加载到内存/磁盘中，然后才会执行分组。我认为目前 spark/ES 连接器不会在 ES 查询语言中翻译 sql 语法。