文章/答案/技术大牛

发布

社区首页 >问答首页 >使用火花从弹性搜索中获取最新的N条记录

问使用火花从弹性搜索中获取最新的N条记录
EN

Stack Overflow用户

提问于 2019-07-22 06:31:13

回答 1查看 333关注 0票数 2

我想检索插入到Elasticsearch中的最后50条记录，以找出异常检测项目的平均值。我就是这样从ES中检索数据的。然而，它正在获取整个数据，而不是最后的50条记录。有什么办法吗？

edf = spark \
   .read \
   .format("org.elasticsearch.spark.sql") \
   .option("es.read.metadata", "false") \
   .option("es.nodes.wan.only","true") \
   .option("es.port","9200")\
   .option("es.net.ssl","false")\
   .option("es.nodes", "http://localhost") \
   .load("anomaly_detection/data")

# GroupBy based on the `sender` column
df3 = edf.groupBy("sender") \
     .agg(expr("avg(amount)").alias("avg_amount"))

在这里，sender列正在获取整个行数据，如何只获取最后50个DataFrame行数据？

输入数据模式格式：

|sender|receiver|amount|

pyspark

spark-structured-streaming

apache-spark

elasticsearch

回答 1

Stack Overflow用户

回答已采纳

发布于 2019-07-22 06:40:55

您还可以在读取数据时将查询添加为

query='{"query": {"match_all": {}}, "size": 50, "sort": [{"_timestamp": {"order": "desc"}}]}'

把它当作

edf = spark \
   .read \
   .format("org.elasticsearch.spark.sql") \
   .option("es.read.metadata", "false") \
   .option("es.nodes.wan.only","true") \
   .option("es.port","9200")\
   .option("es.net.ssl","false")\
   .option("es.nodes", "http://localhost") \
   .option("query", query)
   .load("anomaly_detection/data")

票数 1

页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持

原文链接：

https://stackoverflow.com/questions/57140273

复制

相似问题

问使用火花从弹性搜索中获取最新的N条记录
EN

回答 1

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问使用火花从弹性搜索中获取最新的N条记录EN

回答 1

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问使用火花从弹性搜索中获取最新的N条记录
EN