taosiyuan163 commented on code in PR #132: URL: https://github.com/apache/flink-ml/pull/132#discussion_r930911825
########## flink-ml-lib/src/main/java/org/apache/flink/ml/stats/chisqtest/ChiSqTest.java: ########## @@ -0,0 +1,492 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.ml.stats.chisqtest; + +import org.apache.flink.api.common.functions.MapPartitionFunction; +import org.apache.flink.api.common.functions.RichFlatMapFunction; +import org.apache.flink.api.common.functions.RichMapFunction; +import org.apache.flink.api.common.functions.RichMapPartitionFunction; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.api.java.tuple.Tuple3; +import org.apache.flink.api.java.tuple.Tuple4; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.ml.api.AlgoOperator; +import org.apache.flink.ml.common.broadcast.BroadcastUtils; +import org.apache.flink.ml.common.datastream.DataStreamUtils; +import org.apache.flink.ml.param.Param; +import org.apache.flink.ml.util.ParamUtils; +import org.apache.flink.ml.util.ReadWriteUtils; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.BoundedOneInput; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.api.internal.TableImpl; +import org.apache.flink.types.Row; +import org.apache.flink.util.Collector; +import org.apache.flink.util.Preconditions; + +import org.apache.commons.math3.distribution.ChiSquaredDistribution; + +import java.io.IOException; +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * Chi-square test of independence of variables in a contingency table. This Transformer computes + * the chi-square statistic,p-value,and dof(number of degrees of freedom) for every feature in the + * contingency table, which constructed from the `observed` for each categorical values. All label + * and feature values must be categorical. + * + * <p>See: http://en.wikipedia.org/wiki/Chi-squared_test. + */ +public class ChiSqTest implements AlgoOperator<ChiSqTest>, ChiSqTestParams<ChiSqTest> { + + final String bcCategoricalMarginsKey = "bcCategoricalMarginsKey"; + final String bcLabelMarginsKey = "bcLabelMarginsKey"; + + private final Map<Param<?>, Object> paramMap = new HashMap<>(); + + public ChiSqTest() { + ParamUtils.initializeMapWithDefaultValues(paramMap, this); + } + + @Override + public Table[] transform(Table... inputs) { + + final String[] inputCols = getInputCols(); + String labelCol = getLabelCol(); + + StreamTableEnvironment tEnv = + (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment(); + + SingleOutputStreamOperator<Tuple3<String, Object, Object>> colAndFeatureAndLabel = + tEnv.toDataStream(inputs[0]) + .flatMap(new ExtractColAndFeatureAndLabel(inputCols, labelCol)); + + // compute the observed frequencies + DataStream<Tuple4<String, Object, Object, Long>> observedFreq = + DataStreamUtils.mapPartition( + colAndFeatureAndLabel.keyBy(Tuple3::hashCode), + new GenerateObservedFrequencies()); + + SingleOutputStreamOperator<Tuple4<String, Object, Object, Long>> filledObservedFreq = + observedFreq + .transform( + "filledObservedFreq", + Types.TUPLE( + Types.STRING, + Types.GENERIC(Object.class), + Types.GENERIC(Object.class), + Types.LONG), + new FillZeroFunc()) + .setParallelism(1); + + // return a DataStream of the marginal sums of the factors + DataStream<Tuple3<String, Object, Long>> categoricalMargins = + DataStreamUtils.mapPartition( + observedFreq.keyBy(tuple -> new Tuple2<>(tuple.f0, tuple.f1).hashCode()), + new MapPartitionFunction< + Tuple4<String, Object, Object, Long>, + Tuple3<String, Object, Long>>() { + @Override + public void mapPartition( + Iterable<Tuple4<String, Object, Object, Long>> iterable, + Collector<Tuple3<String, Object, Long>> out) { + HashMap<Tuple2<String, Object>, Long> map = new HashMap<>(); + + for (Tuple4<String, Object, Object, Long> tuple : iterable) { + Long observedFreq = tuple.f3; + Tuple2<String, Object> key = new Tuple2<>(tuple.f0, tuple.f1); + + if (map.containsKey(key)) { + Long count = map.get(key); + map.put(key, count + observedFreq); + } else { + map.put(key, observedFreq); + } + } + + for (Tuple2<String, Object> key : map.keySet()) { + Long categoricalMargin = map.get(key); + out.collect(new Tuple3<>(key.f0, key.f1, categoricalMargin)); + } + } + }); + + // return a DataStream of the marginal sums of the labels + DataStream<Tuple3<String, Object, Long>> labelMargins = Review Comment: As above, I used `transform()` and `BoundedOneInput#endInput()` to optimize the implementation. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
