1

我正在编写一个 spark 应用程序，用于在一个时间范围内查找前 n 个访问的 URL。但是这项工作一直在运行，并且需要几个小时才能389451在 ES 中记录一个实例。我想减少这个时间。

我正在阅读火花中的弹性搜索，如下所示

 val df = sparkSession.read
    .format("org.elasticsearch.spark.sql")
    .load(date + "/" + business)
    .withColumn("ts_str", date_format($"ts", "yyyy-MM-dd HH:mm:ss")).drop("ts").withColumnRenamed("ts_str", "ts")
    .select(selects.head, selects.tail:_*)
    .filter($"ts" === ts)
    .withColumn("url", split($"uri", "\\?")(0)).drop("uri").withColumnRenamed("url", "uri").cache()

在上面的 DF 中，我正在从 ElasticSearch 中读取和过滤。此外，我正在从 URI 中删除查询参数。

然后我正在分组

var finalDF = df.groupBy("col1","col2","col3","col4","col5","uri").agg(sum("total_bytes").alias("total_bytes"), sum("total_req").alias("total_req"))

然后我正在运行一个窗口函数

val partitionBy = Seq("col1","col2","col3","col4","col5")

val window = Window.partitionBy(partitionBy.head, partitionBy.tail:_*).orderBy(desc("total_req"))


finalDF = finalDF.withColumn("rank", rank.over(window)).where($"rank" <= 5).drop("rank")

然后我正在写 finalDF 到 cassandra

finalDF.write.format("org.apache.spark.sql.cassandra").options(Map("table" -> "table", "keyspace" -> "keyspace")).mode(SaveMode.Append).save()

我在 ES 集群中有 4 个数据节点，我的 Spark 机器是 16 核 64GB Ram VM。请帮我找出问题所在。

4

1 回答 1

-1

在读取后保留数据框可能是一个好主意，因为您将在 rank 函数中使用这么多次。

于 2018-01-05T09:58:02.193 回答