- iOS/Objective-C 元类和类别
- objective-c - -1001 错误,当 NSURLSession 通过 httpproxy 和/etc/hosts
- java - 使用网络类获取 url 地址
- ios - 推送通知中不播放声音
如何使用 Apache Beam 框架按降序排序?
我设法创建了一个字数统计管道,它按字词的字母顺序对输出进行排序,但没有弄清楚如何反转排序顺序。
代码如下:
public class SortedWordCount {
public static void main(String[] args) {
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
BufferedExternalSorter.Options options1 = BufferedExternalSorter.options();
p.apply(TextIO.read().from("d:/dev/playground/apache/beam/word-count-beam/src/test/resources/bible/whole_bible.txt"))
.apply("ExtractWords", ParDo.of(new DoFn<String, String>() {
@ProcessElement
public void processElement(ProcessContext c) {
for (String word : c.element().split(ExampleUtils.TOKENIZER_PATTERN)) {
if (!word.isEmpty()) {
c.output(word);
}
}
}
}))
.apply(Count.perElement())
.apply(ParDo.of(new DoFn<KV<String, Long>, KV<String, Long>>() {
@ProcessElement
public void processElement(ProcessContext c){
KV<String, Long> element = c.element();
if(element.getKey().length() > 2) {
c.output(element);
}
}
}))
.apply("CreateKey", MapElements.via(new SimpleFunction<KV<String, Long>, KV<String, KV<String, Long>>>() {
public KV<String, KV<String, Long>> apply(KV<String, Long> input) {
return KV.of("sort", KV.of(input.getKey().toLowerCase(), input.getValue()));
}
}))
.apply(GroupByKey.create())
.apply(SortValues.create(options1))
.apply("FormatResults", MapElements.via(new SimpleFunction<KV<String, Iterable<KV<String, Long>>>, String>() {
@Override
public String apply(KV<String, Iterable<KV<String, Long>>> input) {
return StreamSupport.stream(input.getValue().spliterator(), false)
.map(value -> String.format("%20s: %s", value.getKey(), value.getValue()))
.collect(Collectors.joining(String.format("%n")));
}
}))
.apply(TextIO.write().to("bible"));
// Run the pipeline.
p.run().waitUntilFinish();
}
}
此代码生成一个按字母顺序排序的单词列表及其各自的计数:
aaron: 350
aaronites: 2
abaddon: 1
abagtha: 1
abana: 1
abarim: 4
abase: 4
abased: 4
abasing: 1
abated: 6
abba: 3
abda: 2
abdeel: 1
abdi: 3
abdiel: 1
abdon: 8
abednego: 15
abel: 16
abelbethmaachah: 2
abelmaim: 1
经过一些调试我知道代码使用类:
org.apache.beam.sdk.extensions.sorter.InMemorySorter
此类在执行排序方法期间使用静态最终比较器:
private static final Comparator<byte[]> COMPARATOR = UnsignedBytes.lexicographicalComparator();
public Iterable<KV<byte[], byte[]>> sort() {
checkState(!sortCalled, "sort() can only be called once.");
sortCalled = true;
Comparator<KV<byte[], byte[]>> kvComparator =
new Comparator<KV<byte[], byte[]>>() {
@Override
public int compare(KV<byte[], byte[]> o1, KV<byte[], byte[]> o2) {
return COMPARATOR.compare(o1.getKey(), o2.getKey());
}
};
Collections.sort(records, kvComparator);
return Collections.unmodifiableList(records);
}
无法在此类中注入(inject)比较器。
最佳答案
我最终听从了 jkff 的建议。并使用 Apache Beam 重写了小型 WordCount。我还摆脱了 SortValues
,只是将记录分组到一个键中,然后自己进行排序。
这是我想出的:
import org.apache.beam.examples.common.ExampleUtils;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.*;
import org.apache.beam.sdk.values.KV;
import java.util.ArrayList;
import java.util.function.Supplier;
import java.util.stream.StreamSupport;
public class DescendingWordCount {
public static void main(String[] args) {
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
p.apply(TextIO.read().from("d:/whole_bible.txt"))
.apply("ExtractWords", ParDo.of(new DoFn<String, String>() {
@ProcessElement
public void processElement(ProcessContext c) {
for (String word : c.element().split(ExampleUtils.TOKENIZER_PATTERN)) {
if (word.length() > 1) {
c.output(word.toLowerCase());
}
}
}
}))
.apply(Count.perElement())
.apply("CreateKey", ParDo.of(new DoFn<KV<String, Long>, KV<String, KV<String, Long>>>() {
@ProcessElement
public void processElement(ProcessContext c) {
KV<String, Long> element = c.element();
String key = element.getKey();
c.output(KV.of("single", KV.of(key, element.getValue())));
}
}))
.apply(GroupByKey.create())
.apply("FormatResults",
MapElements.via(
new SimpleFunction<KV<String, Iterable<KV<String, Long>>>, String>() {
@Override
public String apply(KV<String, Iterable<KV<String, Long>>> input) {
return StreamSupport.stream(input.getValue().spliterator(), false)
.collect((Supplier<ArrayList<KV<String, Long>>>) ArrayList::new,
(al, kv) -> al.add(KV.of(kv.getKey(), kv.getValue())),
(sb, kv) -> {
})
.stream()
.sorted((kv1, kv2) -> kv2.getKey().compareTo(kv1.getKey()))
.collect(StringBuilder::new,
(sb, kv) -> sb.append(String.format("%20s : %d%n", kv.getKey(), kv.getValue())),
(sb, kv) -> {
}).toString();
}
}
))
.apply(TextIO.write().withNumShards(1).to("minimal-wordcount-bible"));
p.run().waitUntilFinish();
}
}
这会打印出如下输出:
zuzims : 1
zurishaddai : 5
zuriel : 1
zur : 5
zuph : 3
zuar : 5
zorobabel : 3
zorites : 1
zoreah : 1
zorathites : 1
zorah : 8
zophim : 1
zophar : 4
zophai : 1
zophah : 2
zoheth : 1
zoheleth : 1
zohar : 4
zobebah : 1
zobah : 11
zoba : 2
zoar : 10
zoan : 7
zizah : 1
ziza : 2
ziz : 1
zithri : 1
zipporah : 3
关于java - Apache Beam 框架 - 按降序排序,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/47760301/
我正在尝试对每个条目有多个值的关联数组进行排序。 例如 [0] => stdClass Object ( [type] => node [sid] => 158 [score] => 0.059600
我在 mysql 中有“日期”列以这种格式保存日期 2014 年 9 月 17 日(日-月-年) 我需要对它们进行升序排序,所以我使用了这个命令: SELECT * FROM table ORDER
我目前正在将 MySQL 存储过程重写为 MS SQL 存储过程,但遇到了问题。 在 MySQL 存储过程中,有一个游标,它根据最近的日期 (effdate) 选择一个值并将其放入变量 (thestt
我想要 gwt r.QuestionId- 排序。但是我得到未排序的 QuestionId 尽管我提到了 QuestionId ASC 的顺序。 SELECT r.QuestionId,
我有一个关于在 scandir 函数中排序的基本问题。到目前为止,我阅读了 POSIX readdir 的手册页,但没有找到有关订购保证的具体信息。 但是当我遍历大目录(无法更改,只读)时,我在多个系
基本上我必须从 SQL 数据库中构建项目列表,但是用户可以选择对 7 个过滤器的任意组合进行过滤,也可以选择要排序的列以及按方向排序。 正如您可以想象的那样,这会以大量不同的组合进行编码,并且数据集非
我有两张 table 。想象第一个是一个目录,包含很多文件(第二个表)。 第二个表(文件)包含修改日期。 现在,我想选择所有目录并按修改日期 ASC 对它们进行排序(因此,最新的修改最上面)。我不想显
我想先根据用户的状态然后根据用户名来排序我的 sql 请求。该状态由 user_type 列设置: 1=活跃,2=不活跃,3=创始人。 我会使用此请求来执行此操作,但它不起作用,因为我想在“活跃”成员
在 C++ 中,我必须实现一个“类似 Excel/Access”(引用)的查询生成器,以允许对数据集进行自定义排序。如果您在 Excel 中使用查询构建器或 SQL 中的“ORDER BY a, b,
我面临这样的挑战: 检索按字段 A 排序的文档 如果字段 B 存在/不为空 . 否则 按字段排序 C. 在 SQL 世界中,我会做两个查询并创建一个 UNION SELECT,但我不知道如何从 Mon
我想对源列表执行以下操作: map 列表 排序 折叠 排序 展开 列表 其中一些方法(例如map和toList)是可链接的,因为它们返回非空对象。但是,sort 方法返回 void,因为它对 List
我制作了一个用于分析 Windows 日志消息编号的脚本。 uniq -c 数字的输出很难预测,因为根据数字的大小会有不同的空白。此时,我手动删除了空白。 这是对消息进行排序和计数的命令: cat n
我有以下词典: mydict1 = {1: 11, 2: 4, 5: 1, 6: 1} mydict2 = {1: 1, 5: 1} 对于它们中的每一个,我想首先按值(降序)排序,然后按键(升序)排序
我刚刚开始使用泛型,目前在对多个字段进行排序时遇到问题。 案例: 我有一个 PeopleList 作为 TObjectList我希望能够通过一次选择一个排序字段,但尽可能保留以前的排序来制作类似 Ex
有没有办法在 sql 中组合 ORDER BY 和 IS NULL 以便我可以在列不为空时按列排序,但如果它为null,按另一列排序? 最佳答案 类似于: ORDER BY CASE WHEN
我有一个包含 2 列“id”和“name”的表。 id 是常规的自动增量索引,name 只是 varchar。 id name 1 john 2 mary 3 pop 4 mary 5 j
场景 网站页面有一个带有分页、过滤、排序功能的表格 View 。 表中的数据是从REST API服务器获取的,数据包含数百万条记录。 数据库 REST API 服务器 Web 服务器 浏览器 问
假设我有一本字典,其中的键(单词)和值(分数)如下: GOD 8 DONG 16 DOG 8 XI 21 我想创建一个字典键(单词)的 NSArray,首先按分数排序,然后按字
如何在 sphinx 上通过 sql 命令选择前 20 行按标题 WEIGHT 排序,接下来 20 行按标题 ASC 排序(总共 40 个结果),但不要给出重复的标题输出。 我尝试了这个 sql 命令
我有一个奇怪的问题,当从 SQLite 数据库中选择信息并根据日期排序时,返回的结果无效。 我的SQL语句是这样的: Select pk from usersDates order by dateti
我是一名优秀的程序员,十分优秀!