当前位置: 首页 > 知识库问答 >
问题:

沉入爪哇列表可能与榛子喷气机?

闾丘博超
2023-03-14

我有一个帐户列表,并在滴答上执行hashjoin,返回带有滴答数据的帐户。但是在hashjoin之后,我将导入llistjet,然后用distributedstream读取并返回它。

public List<Account> populateTicksInAccounts(List<Account> accounts) {
    ...
    ...
    Pipeline p = Pipeline.create();
    BatchSource<Tick> ticksSource = Sources.list(TICKS_LIST_NAME);
    BatchSource<Account> accountSource = Sources.fromProcessor(AccountProcessor.of(accounts));

    p.drawFrom(ticksSource)
        .hashJoin(p.drawFrom(accountSource), JoinClause.joinMapEntries(Tick::getTicker), accountMapper())
        .drainTo(Sinks.list(TEMP_LIST));

    jet.newJob(p).join();
    IListJet<Account> list = jet.getList(TEMP_LIST);
    return DistributedStream.fromList(list).collect(DistributedCollectors.toIList());
}

在执行hashjoin之后,是否可以drainteto javalist而不是llistjet

下面这样的事情是可能的?

IListJet<Account> accountWithTicks = new ArrayList<>();
p.drawFrom(ticksSource)
    .hashJoin(p.drawFrom(accountSource), JoinClause.joinMapEntries(Tick::getTicker), accountMapper())
    .drainTo(<CustomSinkProcessor(accountWithTicks)>);
return accountWithTicks;

CustomSinkProcessor中的哪里会拿空的java列表并带着帐户返回?

共有1个答案

周朗
2023-03-14

请记住,您提交给Jet执行的代码在您提交代码的进程之外运行。虽然理论上可以提供您所要求的API,但实际上它只需要执行一些技巧来在集群的每个成员上运行代码,让所有成员将其结果发送到一个位置,并填充一个列表返回给您。这将违背分布式计算的本质。

如果您认为这将有助于代码的可读性,可以编写如下的帮助器方法:

public <T, R> List<R> drainToList(GeneralStage<T> stage) {
    String tmpListName = randomListName();
    SinkStage sinkStage = stage.drainTo(Sinks.list(tmpListName));
    IListJet<R> tmpList = jet.getList(tmpListName);
    try {
        jet.newJob(sinkStage.getPipeline()).join();
        return new ArrayList<>(tmpList);
    } finally {
        tmpList.destroy();
    }
}

特别注意一行

return new ArrayList<>(tmpList);

与您的

IListJet<Account> list = jet.getList(TEMP_LIST);
return DistributedStream.fromList(list).collect(DistributedCollectors.toIList());

这只是将一个Hazelcast列表复制到另一个列表,并返回它的句柄。现在您已经泄露了Jet集群中的两个列表。当你停止使用它们时,它们不会自动消失。

 类似资料:
  • 问题内容: 给定以下声明(取自该 Oracle Java教程),该声明与Collections类的binarySearch()方法有关: 两种形式的返回值都相同。如果列表包含搜索关键字,则返回其索引。如果不是,则返回值为 (-(插入点)-1) ,其中插入点是将值插入到List中的点,或者第一个元素的索引大于该值或list.size。 (),如果List中的所有元素都小于指定的值。 为什么返回值no

  • 程序在最后未能计算出正确的和。我做错了什么?如有任何帮助,不胜感激。谢了。

  • 我试图在macOS高Sierra上的eclipse氧气4.7.3a上使用windows Builder。它安装良好,但每次我尝试打开设计部分时,它都会显示此错误 Eclipse在0下运行,但是这个Java项目有1.8Java的合规级别,因此WindowBuilder将无法从该项目加载类。对项目使用较低级别的Java,或者使用较新的Java版本运行Eclipse。 我什么都试过了。就像改变运行环境,

  • 我是jetpack compose的新手,从< code>JetpackCompose基础代码中学习它,所以我在创建一个高性能的懒惰列表中找到了它 注意:LazyColumn不会像那样回收其子项。当您滚动浏览它时,它会发出新的,并且仍然具有高性能,因为与实例化Android视图相比,发出相对便宜。 因此,发出新的可组合项是多么的便宜和高效,那么为什么不使用列并一次性组合整个列表,而不是在滚动时不断

  • 问题内容: 我有这个Java代码段。我是Java的菜鸟。 错误: 码: 问题答案: 是的,这是问题所在: 在课程的最高级别,您只能拥有: 实例初始化程序块() 静态初始值设定块() 变量声明 构造函数声明 方法声明 嵌套类型声明 终结器声明 这些都不是。如果你 的意思 来声明一个变量,你应该这样做: 如果这 不是 您想要的,则应解释您的意图。 编辑:修复此问题后,此编译器错误似乎很明显: Conf

  • 我已经红色了多篇文章和讨论,但我仍然有一些不确定性:我不确定是否应该使用或任何其他类型来存储预订-在“在线预订”的意义上(因此来自不同国家/时区的参与者需要在时间线上的同一时刻会面)。我倾向于使用,因为DB和Backend设置为UTC,并且由于传入的“创建预订”json消息包含ISO 8601(带偏移量)start DateTime 让我们采用以下设置:1.数据库(UTC、Oracle、MSSQL