Stream.reduceのための数学

３行で

• java.util.stream.Stream#reduce
• java.util.function.Function#identity

はどちらもとても出来る子です。（ただし後者は報われない）

Optional<T> reduce(BinaryOperator<T> accumulator)

必要になる数学用語

元(element)

ある集合(クラス)の要素を元と呼ぶ。
Javaではクラス X の実体(インスタンス/オブジェクト) x になる。
誰が訳したのか知らないけど英語とのとっつきやすさの差は一体何なんでしょうね。

項

だいたい引数と同義。

二項関数

引数を２つとる関数。
Javaでは BiFunction<T,U,R>になる。

二項演算

引数を２つとる関数のうち、全ての項と結果が同じ集合の元となるもの。
Javaでは BinaryOperator<T> になる。

二項演算 f とその対象となるクラス T をペアにしたもの (T,f) をマグマと呼ぶ。
煮えてそうな名前だけどこの段階では煮ても焼いても食えない。

結合法則

二項演算が持っている__かもしれない__性質の名前。
次の性質を持つ二項演算(BinaryOperator<T>)は__結合的二項演算__と呼ぶ。

f.apply(t1, f.apply(t2, t3)) = f.apply(f.apply(t1, t2), t3)

ちなみにJavadocにも同じことが書かれている。

あたり前じゃないかって？ とんでもない。

引き算は__二項演算だけど結合的ではない。__

$(1 - (-2)) - 3 = 3 - 3 = 0$
$1 - ((-2) - 3) = 1 - (-5) = 6$

割り算も__二項演算だけど結合的ではない。__(ついでにゼロ除算という地雷もある)

$20 / 2 / 2 = 10 / 2 = 5$
$20 / (2 / 2) = 20 / 1 = 20$

ちなみに__足し算と掛け算は結合的__だ。検算は自分でしてください。
こういう結合的二項演算とその対象になるクラス T のマグマは半群と呼ぶ。

『簡単に足し算と掛け算に変換できるじゃないか』というのはちょっと待って欲しい。そいつはおおとりだ。

なぜ結合的である必要があるのか

accumulator - 2つの値を結合するための結合的、非干渉、およびステートレスな関数

Streamでの処理は並列実行を前提としているからだ。
Stream.of(t1, t2, t3, t4, t5) に対して accumulator はこういうふうに処理される可能性がある。

• スレッド1: accumulator.apply(t1, t2)
• スレッド2: accumulator.apply(t3, t4)
• 合流後のメインスレッド: accumulator.apply(accumulator.apply(スレッド1結果, スレッド2結果), t5)

具体例いってみよう。

import java.math.BigDecimal;
import java.util.stream.Stream;
import java.util.stream.IntStream;
import java.util.function.Supplier;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;

interface Reduction1 {
	static void main(String[] args) {
		Supplier<Stream<BigDecimal>> s =
      		() -> IntStream.of(
        	100000, 10000, 1000, 100, 10, 1
      	).mapToObj(BigDecimal::valueOf);
      	BiConsumer<
      		Supplier<Stream<BigDecimal>>,
       		BinaryOperator<BigDecimal>
       	> each = (x,op) -> {
       		System.out.print("直列:");
       		x.get().reduce(op).ifPresent(System.out::println);
       		System.out.print("並列:");
       		x.get().parallel().reduce(op).ifPresent(System.out::println);
       	};
       	each.accept(s, (x,y) -> x.add(y));
       	each.accept(s, (x,y) -> x.multiply(y));
       	each.accept(s, (x,y) -> x.subtract(y));
       	each.accept(s, (x,y) -> x.divide(y));
    }
}

直列:111111
並列:111111
直列:1000000000000000
並列:1000000000000000
直列:88889
並列:90909
直列:0.00001
並列:1000

非結合的な引き算と割り算では直列実行時と並列実行時で異なる結果がでてしまう。

ポイント

reduce(accumulator) に使う accumulator が結合的であることは使う側が保証しなくてはならない。

T reduce(T identity, BinaryOperator<T> accumulator)

必要になる数学用語

単位元(identity element)

マグマ(T,f)に存在する__かもしれない__元の名前。
ある元 t0 と__それ以外の全ての元__ (tx) で次が成立する場合、t0 を二項演算(BinaryOperator<T>) f の__単位元__と呼ぶ。二項演算のほうは__単位的である__と表現する。

f.apply(t0, tx) = tx
f.apply(tx, t0) = tx

単位元はマグマ(T,f)に__存在しない__か__１つだけ存在するか__のどちらかしかありえない。
数学的な言い方だと「高々一つしか存在しない」。慣れの問題だろうけどわかんねーよ。
仮に元 ta, tb で次が成立するとして

f.apply(ta, tx) = tx
f.apply(tx, ta) = tx
f.apply(tb, tx) = tx
f.apply(tx, tb) = tx

この結果からどちらかが単位元か、単位元がそもそも存在しないかがわかるからだ。

f.apply(ta, tb) = ?
f.apply(tb, ta) = ?

• 両方とも tb なら単位元は ta
• 両方とも ta なら単位元は tb
• それ以外なら__単位元は存在しない__

半群のうち単位元のあるものはモノイドと呼ぶ。
モナモナ言わないで欲しい。名前は似てるけどモナドとは別物だよ！怖くない！
ちなみにJavadocには単位元の詳細な説明がない。

なぜ単位的だと嬉しいのか。

identity - 蓄積関数に対する単位元の値

Optional<T> reduce(BinaryOperator<T> accumulator)の存在からわかるように結合的二項演算さえあれば別に__なくても困らない__子だ。

前掲の例でいうと Stream.of(t1, t2, t3, t4, t5) に対して accumulator はこういうふうに処理される可能性がある。

• スレッド1: accumulator.apply(t1, t2)
• スレッド2: accumulator.apply(t3, t4)
• スレッド3: accumulator.apply(t0, t5)
• 合流後のメインスレッド: accumulator.apply(accumulator.apply(スレッド1結果, スレッド2結果), スレッド3結果)

単位元 t0 は__演算結果に影響を与えない__ので対象が奇数個でも並列計算ができるんだ！やったね！

具体例いってみよう。

足し算の単位元は0、掛け算の単位元は1になる。検算は例によって自ｂ

import java.math.BigDecimal;
import java.util.stream.Stream;
import java.util.stream.IntStream;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;

interface Reduction2 {
	static void main(String[] args) {
		Supplier<Stream<BigDecimal>> s =
			() -> IntStream.of(
			100000, 10000, 1000, 100, 10, 1
		).mapToObj(BigDecimal::valueOf);
		Function<BigDecimal, BiConsumer<
		Supplier<Stream<BigDecimal>>,
		BinaryOperator<BigDecimal>
		>> each = i -> (x,op) -> {
			System.out.print("直列:");
			System.out.println(x.get().reduce(i,op));
			System.out.print("並列:");
			System.out.println(x.get().parallel().reduce(i,op));
		};
		each.apply(BigDecimal.valueOf(0)).accept(s, (x,y) -> x.add(y));
		each.apply(BigDecimal.valueOf(1)).accept(s, (x,y) -> x.multiply(y));
		each.apply(BigDecimal.valueOf(1)).accept(s, (x,y) -> x.add(y));
		each.apply(BigDecimal.valueOf(2)).accept(s, (x,y) -> x.multiply(y));
	}
}

直列:111111
並列:111111
直列:1000000000000000
並列:1000000000000000
直列:111112
並列:111117
直列:2000000000000000
並列:64000000000000000

第1引数に単位元以外を使うと直列実行時と並列実行時で異なる結果がでてしまう。

ポイント

reduce(identity, accumulator) に使う identity が単位元であることは使う側が保証しなくてはならない。

<U> U reduce(U identity, BiFunction<U,? super T,U> accumulator, BinaryOperator<U> combiner)

ここまでは前座である。ここから何だかモナモナ言う声が聞こえてくる。怖い。
Javadocも何か混乱している。

accumulator - 追加の要素を結果に組み込むための、結合的、非干渉およびステートレスな関数

Javadocの説明はこうなっているがそもそも accumulator は二項演算ではなく二項関数なので結合法則の成立のしようが無いです。

combiner - 2つの値を結合するための結合的、非干渉およびステートレスな関数(アキュムレータ関数と互換性がなければいけない)

互換性だと何言っているのかよくわかりません。

identity値はコンバイナ関数の単位元でなければいけません。つまり、すべてのuについて、combiner(identity, u)がuに等しくなります。さらに、combiner関数はaccumulator関数と互換性がある必要があります。すべてのuとtについて、次が成り立つ必要があります。

combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)

どうやらこれが全てのようです。これについての数学的なバックエンドとかは__ごめんなさい、よくわかりません__。

ポイント

3項reduceの結果は以下の形式になる。

g(f(f(f(f(u, t[0]), t[1]), t[2]), t[3]..., t[n]), f(f(f(f(0, t[n+1]), t[n+2]), t[n+3]), t[n+4]))

• u = identity
• f = accumulator
• g = combiner

使う側は以下が成立することを保証する必要がある。

g(u, f(u, t[x])) = f(u, t[x])

並列性実行可能性を担保したうえで使うのは__かなり難しい__。

必要かどうかはよく考えよう

APIの注:
この形式を使用するリダクションの多くは、map操作とreduce操作を明示的に組み合わせることで、より単純に表現することができます。accumulator関数はマッパーとアキュムレータを融合したものとして動作しますが、これは、以前のリデュース値がわかっていれば一部の計算を回避できる場合など、マッピングとリダクションを個別に実行するよりも効率的になる場合があります。

何も考えずに reduce(identity, accumulator, combiner)しがちなので耳に痛いです。

必要になる数学用語

マグマの準同型

あるマグマ(T,f)と別のマグマ(U,g)の間に次の関係を成立させる単項関数(参照先では写像)$h(tx) = gx$を準同型と呼ぶ。

$h(f(t1,t2)) = g(h(t1),h(t2))$

『簡単に足し算と掛け算に変換できるじゃないか』という話はこれと同じことを言っている事になる。怖い。

引き算から足し算への準同型は $h(x)=0-x$ で割り算から掛け算掛け算から割り算への順同型は $h(x)=1/x$ になる。

この形式を使用するリダクションの多くは、map操作とreduce操作を明示的に組み合わせることで、より単純に表現することができます。

これが成り立つパターンの accumulator はクラスUの結合的二項演算

g(u1,u2) = u3

にクラスTからクラスU への順同型変換

h(t)= u

をこういう感じに合成したものになります。

accumulator(u,t)=g(u,h(t))

要は引き算の場合は map(x -> -1) の後に足し算、割り算の場合は map(x -> 1 / x) した後に掛け算しろよ、って話になります。

よく考えなかった例1

引き算、割り算では前掲の通りmapしろよという話になってしまうのでここでは逆順Streamを構成します。

import java.util.stream.Stream;
import java.util.function.Function;

interface Reduction3 {
	static void main(String[] args) {
		Function<
			Stream.Builder<Integer>,
			Stream.Builder<Integer>> reversed =
			Stream.of(1,2,3,4,5).reduce(
			Function.identity(),
			(f, x) -> f.compose(xs -> xs.add(x)),
			(f, g) -> f.compose(g));
		reversed.apply(
		reversed.apply(Stream.builder()))
			.build().forEach(System.out::print);
	}
}

5432154321

関数合成によって reversed に xs -> xs.add(5).add(4).add(3).add(2).add(1) という処理が__畳み込まれている__のと__Function.identity()__の活躍に注目！
そう！ Function.identity() は__関数合成の単位元__だったのです！！

import java.util.stream.Stream;
import java.util.function.Function;

interface Reduction4 {
	static void main(String[] args) {
		Function<
			Stream.Builder<Integer>,
			Stream.Builder<Integer>> reversed =
			Stream.of(1,2,3,4,5).reduce(
			x -> x,
			(f, x) -> f.compose(xs -> xs.add(x)),
			(f, g) -> f.compose(g));
		reversed.apply(
		reversed.apply(Stream.builder()))
			.build().forEach(System.out::print);
	}
}

ただし x -> x のほうが短いという報われない子です。

##準同型

前掲の具体例は準同型がありました。

import java.util.stream.Stream;
import java.util.function.Function;

interface Reduction5 {
    static void main(String[] args) {
        Function<
            Stream.Builder<Integer>,
            Stream.Builder<Integer>> reversed =
            Stream.of(1,2,3,4,5).<Function<
            	Stream.Builder<Integer>,
                Stream.Builder<Integer>
            >>map(x -> xs -> xs.add(x))
            .reduce(x -> x, (f, g) -> f.compose(g));
        reversed.apply(
        reversed.apply(Stream.builder()))
            .build().forEach(System.out::print);
    }
}

よく考えなかった例2

この記事で課題になっている素因数分解だ。
チャーチ対を使って並列計算の整合性も保っています。

import java.util.stream.Stream;
import java.util.stream.IntStream;
import java.util.function.Function;

interface Reduction6 {
    static void main(String[] args) {
        Function<Integer, Stream<Integer>> func = num ->
            IntStream.iterate(num, i ->
                i     == 1 ? 1 :
                i % 2 == 0 ? i / 2 :
                             i / IntStream
                                .iterate(3,j->j+2) 
                                .filter(j->i%j==0)
                                .findFirst()
                                .getAsInt())
            .limit((int)(Math.log(num)/Math.log(2))+1)
            .distinct()
            .boxed()
            .parallel()
            .<Function<
                Function<Integer,
                Function<Integer,
                Function<Stream.Builder<Integer>, Stream.Builder<Integer>>
                >>,
                Function<Stream.Builder<Integer>, Stream.Builder<Integer>>
            >>reduce(
                (headFunc) -> headFunc.apply(0).apply(null),
                (crntFunc, crntElem) -> nextFunc ->
                    crntFunc.apply(
                        headElem -> lastElem ->
                        headElem == 0
                            ? nextFunc.apply(crntElem).apply(crntElem)
                            : nextFunc.apply(headElem).apply(crntElem)
                                .compose(s -> s.add(lastElem / crntElem))),
                (prevFunc, nextFunc) -> thenFunc ->
                    prevFunc.apply(prevHead -> prevLast ->
                    nextFunc.apply(nextHead -> nextLast ->
                        nextHead == 0
                            ? thenFunc.apply(prevHead).apply(nextLast)
                            : thenFunc.apply(prevHead).apply(nextLast)
                                .compose(s -> s.add(prevLast / nextHead))
                    ))
            ).apply(headElem -> lastElem -> Function.identity())
             .apply(Stream.builder())
             .build();
        func.apply(1234567890).forEach(System.out::println);
        System.out.println(func.apply(1234567890).reduce(1, (x,y) -> x * y));
    }
}

3
2
5
3803
3607
3
1234567890

##3項reduceはどこにいった。

篩をかけたらどこかにいってしまった。。。

import java.util.*;
import java.util.stream.*;
import java.util.function.*;

interface Factor extends
    Function<Function<Long,
             Function<Long, 
             Function<LongPredicate,
             Function<LongStream.Builder,LongStream.Builder>>>>,
             Function<LongStream.Builder,LongStream.Builder>> {

    public static void main(String[] args) {
        Function<Long, LongStream> factor = num ->
            Stream.<Factor>iterate(
                f -> f.apply(num).apply(2L).apply(v -> v % 2 != 0),
                f -> g -> f.apply(
                     x ->
                     p ->
                     s -> 1 == x ? xs -> xs
                        : x <  (p * p)
                        ? g.apply(x / x)
                           .apply(x)
                           .apply(s)
                           .compose(xs -> xs.add(x))
                        : 0 == (x % p)
                        ? g.apply(x / p)
                           .apply(p)
                           .apply(s)
                           .compose(xs -> xs.add(p))
                        : ((Function<Long, 
                            Function<LongStream.Builder,
                                    LongStream.Builder>>)
                     q -> g.apply(x)
                           .apply(q)
                           .apply(s.and(v -> v % q != 0)))
                           .apply(LongStream.iterate(p + 1, v -> v + 1)
                                            .filter(s)
                                            .findFirst()
                                            .getAsLong())))
            .limit((long) Math.sqrt(num))
            .parallel()
            .reduce(f -> xs -> xs, (f, g) -> g)
            .apply(x -> p -> s -> xs -> xs)
            .apply(LongStream.builder())
            .build();
        factor.apply(1234567L).forEach(System.out::println);
        System.out.println(factor.apply(1234567L).reduce(1, (x, y) -> x * y));
    }
}

127
9721
1234567

まとめ

普通の foldl と普通の foldr ください。