代码之家 › 专栏 › 技术社区 › Alejandro Alcalde

有效计算Flink中的属性数

apache-flink scala

Alejandro Alcalde · 技术社区 · 7 年前

我想知道打火石的属性数量 DataSet 第一次尝试是:

dataset.input
    .map(_.vector.size)
    .reduce((_, b) => b)
    .collect
    .head

然后,看看 Solver 我看到它是这样做的:

// TODO: Faster way to do this?
dataset.map(_.vector.size)
    .reduce((a, b) => b)

但是托多评论说明了一切。

因此,我提出了这个实现:

dataset.first(1)
    .map(_.vector.size)
    .reduce((_, b) => b)
    .collect

是否有更有效的实施?

1 回复 | 直到 7 年前

Fabian Hueske 7 年前

最快的是

dataset
  // only forward first vector of each partition
  .mapPartition(in => if (in.hasNext) Seq(in.next) else Seq())
  // move all remaining vectors to a single partition, compute size of the first and forward it
  .mapPartition(in => if (in.hasNext) Seq(in.next.vector.size) else Seq()).setParallelism(1)
  .collect

使用 reduce 或 groupReduce 效率较低,因为可能会在不首先减少数据的情况下将数据移动到单个计算机,并导致每个输入记录的函数调用。 mapPartition 每个分区调用一次,并且只转发输入迭代器的第一个元素。

推荐文章

Ellen Spertus · 使用Scala访问HBase的推荐方法

7 年前

skrshn · JAVA尝试在Flink中创建Kafka使用者时出现lang.NoClassDefFoundError(CheckpointedRestoring)

7 年前

Daniel · 目录中的Apache Flink流文件

7 年前

Christos Hadjinikolis · Flink:执行管道java时出错。util。同时发生的TimeoutException:期货在[10000毫秒]后超时

7 年前

Mustafa · Apache Flink:自定义InputFormat仅以1的并行度运行

7 年前

Soheil Pourbafrani · Apache Flink-org的类文件。阿帕奇。弗林克。流式处理。api。斯卡拉。找不到数据流

7 年前

Jicaar · 一个卡夫卡制作人可以被多个接收器安全使用吗?

7 年前

Christos Hadjinikolis · Flink JobExecutionException:akka。客户超时

7 年前

Christos Hadjinikolis · Flink IOException:网络缓冲区数量不足

7 年前

Brutal_JL · flink ReduceFunction的默认触发器是什么?

7 年前