You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Fanoid (via GitHub)" <gi...@apache.org> on 2023/02/09 03:34:49 UTC
[GitHub] [flink-ml] Fanoid commented on a diff in pull request #192: [FLINK-30451] Add Estimator and Transformer for Swing
Fanoid commented on code in PR #192:
URL: https://github.com/apache/flink-ml/pull/192#discussion_r1100094424
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null");
+ }
+ return Tuple2.of(
+ ((Number) row.getFieldAs(userCol)).longValue(),
+ ((Number) row.getFieldAs(itemCol)).longValue());
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAllItemsMapState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userItemsMap.entrySet().stream()
+ .forEach(
+ entry -> {
+ Long user = entry.getKey();
+ Map<Long, String> items = entry.getValue();
+ if (items.size() >= minUserItemInteraction
+ && items.size() <= maxUserItemInteraction) {
+ items.entrySet().stream()
+ .forEach(
+ item ->
+ output.collect(
+ new StreamRecord<>(
+ new Tuple3<>(
+ user,
+ item.getKey(),
+ items))));
+ }
+ });
+
+ userAllItemsMapState.clear();
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<Long, Long>> element) {
+ Tuple2<Long, Long> userAndItem = element.getValue();
+ long user = userAndItem.f0;
+ long item = userAndItem.f1;
+ Map<Long, String> items = userItemsMap.get(user);
+
+ if (items == null) {
+ items = new LinkedHashMap();
+ }
+
+ if (items.size() <= maxUserItemInteraction) {
+ items.put(item, null);
+ }
+
+ userItemsMap.put(user, items);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ userAllItemsMapState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "userAllItemsMapState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(userAllItemsMapState, "userAllItemsMapState")
+ .ifPresent(
+ stat -> {
+ userItemsMap = stat;
+ });
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ userAllItemsMapState.update(Collections.singletonList(userItemsMap));
+ }
+ }
+
+ /** Calculates top N similar items of each item. */
+ private static class ComputingSimilarItems extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Tuple3<Long, Long, Map<Long, String>>, Row>,
+ BoundedOneInput {
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+ private Map<Long, Map<Long, String>> itemUsersMap = new HashMap<>();
+ private ListState<Map<Long, Map<Long, String>>> userLocalItemsMapState;
+ private ListState<Map<Long, Map<Long, String>>> itemUsersMapState;
+
+ private final int k;
+ private final int maxUserNumPerItem;
+ private final int alpha1;
+ private final int alpha2;
+ private final double beta;
+
+ private static Character commaDelimiter = ',';
+ private static Character semicolonDelimiter = ';';
+
+ private ComputingSimilarItems(
+ int k, int maxUserNumPerItem, int alpha1, int alpha2, double beta) {
+ this.k = k;
+ this.maxUserNumPerItem = maxUserNumPerItem;
+ this.alpha1 = alpha1;
+ this.alpha2 = alpha2;
+ this.beta = beta;
+ }
+
+ @Override
+ public void endInput() throws Exception {
+
+ Map<Long, Double> userWeights = new HashMap<>(userItemsMap.size());
+ userItemsMap.forEach(
+ (k, v) -> {
+ int count = v.size();
+ userWeights.put(k, calculateWeight(count));
+ });
+
+ for (long mainItem : itemUsersMap.keySet()) {
+ List<Long> userList = sampleUserList(itemUsersMap.get(mainItem), maxUserNumPerItem);
+ HashMap<Long, Double> id2swing = new HashMap<>();
+
+ for (int i = 0; i < userList.size(); i++) {
+ long u = userList.get(i);
+ for (int j = i + 1; j < userList.size(); j++) {
+ long v = userList.get(j);
+ HashSet<Long> interaction =
+ (HashSet<Long>) new HashSet<>(userItemsMap.get(u).keySet()).clone();
+ interaction.retainAll(userItemsMap.get(v).keySet());
+ if (interaction.size() == 0) {
+ continue;
+ }
+ double similarity =
+ (userWeights.get(u)
+ * userWeights.get(v)
+ / (alpha2 + interaction.size()));
+ for (long simItem : interaction) {
+ if (simItem == mainItem) {
+ continue;
+ }
+ double itemSimilarity =
+ id2swing.getOrDefault(simItem, 0.0) + similarity;
+ id2swing.putIfAbsent(simItem, itemSimilarity);
+ }
+ }
+ }
+
+ ArrayList<Tuple2<Long, Double>> itemAndScore = new ArrayList<>();
+ id2swing.forEach((key, value) -> itemAndScore.add(Tuple2.of(key, value)));
+
+ itemAndScore.sort((o1, o2) -> Double.compare(o2.f1, o1.f1));
+
+ if (itemAndScore.size() == 0) {
+ continue;
+ }
+
+ int itemNums = Math.min(k, itemAndScore.size());
+ StringBuilder sbd = new StringBuilder();
Review Comment:
We can use `String#join` or similar methods to make these lines concise, and no need to copy whole string with removing last character.
```
String itemList =
itemAndScore.stream()
.map(d -> "" + d.f0 + commaDelimiter + d.f1)
.collect(Collectors.joining("" + semicolonDelimiter));
```
##########
flink-ml-python/pyflink/ml/lib/recommendation/swing.py:
##########
@@ -0,0 +1,209 @@
+################################################################################
+# 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.
+################################################################################
+import typing
+
+from pyflink.ml.core.param import Param, StringParam, IntParam, FloatParam, ParamValidators
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.param import HasOutputCol
+from pyflink.ml.lib.recommendation.common import JavaRecommendationTransformer
+
+
+class _SwingParams(
+ JavaWithParams,
+ HasOutputCol
+):
+ """
+ Params for :class:`Swing`.
+ """
+
+ USER_COL: Param[str] = StringParam(
+ "user_col",
+ "Name of user column.",
+ "user",
+ ParamValidators.not_null())
+
+ ITEM_COL: Param[str] = StringParam(
+ "item_col",
+ "Name of item column.",
+ "item",
+ ParamValidators.not_null())
+
+ K: Param[int] = IntParam(
+ "k",
+ "The max number of similar items to output for each item.",
+ 100,
+ ParamValidators.gt(0))
+
+ MAX_USER_NUM_PER_ITEM: Param[int] = IntParam(
+ "max_user_num_per_item",
+ "The max number of users that has purchased for each item. If the number of users that have "
+ + "purchased this item is larger than this value, then only maxUserNumPerItem users will "
+ + "be sampled and used in the computation logic.",
+ 1000,
+ ParamValidators.gt(0))
+
+ MIN_USER_BEHAVIOR: Param[int] = IntParam(
+ "min_user_behavior",
+ "The min number of interaction behavior between item and user.",
+ 10,
+ ParamValidators.gt(0))
+
+ MAX_USER_BEHAVIOR: Param[int] = IntParam(
+ "max_user_behavior",
+ "The max number of interaction behavior between item and user."
+ + "The algorithm filters out activate users.",
+ 1000,
+ ParamValidators.gt(0))
+
+ ALPHA1: Param[int] = IntParam(
+ "alpha2",
+ "This parameter is used to calculate weight of each user.",
+ 15,
+ ParamValidators.gt_eq(0))
+
+ ALPHA2: Param[int] = IntParam(
+ "alpha2",
+ "This parameter is used to calculate similarity of users.",
+ 0,
+ ParamValidators.gt_eq(0))
+
+ BETA: Param[float] = FloatParam(
+ "beta",
+ "This parameter is used to calculate weight of each user.",
+ 0.3,
+ ParamValidators.gt_eq(0))
+
+ def __init__(self, java_params):
+ super(_SwingParams, self).__init__(java_params)
+
+ def set_user_col(self, value: str):
+ return typing.cast(_SwingParams, self.set(self.USER_COL, value))
+
+ def get_user_col(self) -> str:
+ return self.get(self.USER_COL)
+
+ def set_item_col(self, value: str):
+ return typing.cast(_SwingParams, self.set(self.ITEM_COL, value))
+
+ def get_item_col(self) -> str:
+ return self.get(self.ITEM_COL)
+
+ def set_k(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.K, value))
+
+ def get_k(self) -> int:
+ return self.get(self.K)
+
+ def set_max_user_num_per_item(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MAX_USER_NUM_PER_ITEM, value))
+
+ def get_max_user_num_per_item(self) -> int:
+ return self.get(self.MAX_USER_NUM_PER_ITEM)
+
+ def set_min_user_behavior(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MIN_USER_BEHAVIOR, value))
+
+ def get_min_user_behavior(self) -> int:
+ return self.get(self.MIN_USER_BEHAVIOR)
+
+ def set_max_user_behavior(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MAX_USER_BEHAVIOR, value))
+
+ def get_max_user_behavior(self) -> int:
+ return self.get(self.MAX_USER_BEHAVIOR)
+
+ def set_alpha1(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.ALPHA1, value))
+
+ def get_alpha1(self) -> int:
+ return self.get(self.ALPHA1)
+
+ def set_alpha2(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.ALPHA2, value))
+
+ def get_alpha2(self) -> int:
+ return self.get(self.ALPHA2)
+
+ def set_beta(self, value: float):
+ return typing.cast(_SwingParams, self.set(self.BETA, value))
+
+ def get_beta(self) -> float:
+ return self.get(self.BETA)
+
+ @property
+ def user_col(self) -> str:
+ return self.get_user_col()
+
+ @property
+ def item_col(self) -> str:
+ return self.get_item_col()
+
+ @property
+ def k(self) -> int:
+ return self.get_k()
+
+ @property
+ def max_user_num_per_item(self) -> int:
+ return self.get_max_user_num_per_item()
+
+ @property
+ def min_user_behavior(self) -> int:
+ return self.get_min_user_behavior()
+
+ @property
+ def max_user_behavior(self) -> int:
+ return self.get_max_user_behavior()
+
+ @property
+ def alpha1(self) -> int:
+ return self.get_alpha1()
+
+ @property
+ def alpha2(self) -> float:
+ return self.get_alpha2()
+
+ @property
+ def beta(self) -> float:
+ return self.get_beta()
+
+
+class Swing(JavaRecommendationTransformer, _SwingParams):
Review Comment:
In Java code, Swing is an implementation of `AlgoOperator`. So I guess here in Python code, Swing should extends `JavaRecommendationAlgoOperator`.
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/recommendation/SwingTest.java:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.recommendation;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.recommendation.swing.Swing;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests {@link Swing}. */
+public class SwingTest {
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+ private Table trainData;
+ RowTypeInfo trainDataTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
+ },
+ new String[] {"user_id", "item_id"});
+ private static final List<Row> trainRows =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(0L, 10L),
+ Row.of(0L, 11L),
+ Row.of(0L, 12L),
+ Row.of(1L, 13L),
+ Row.of(1L, 12L),
+ Row.of(2L, 10L),
+ Row.of(2L, 11L),
+ Row.of(2L, 12L),
+ Row.of(3L, 13L),
+ Row.of(3L, 12L)));
+
+ private static final List<Row> expectedScoreRows =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(10L, "11,0.058845768947156235;12,0.058845768947156235"),
+ Row.of(11L, "10,0.058845768947156235;12,0.058845768947156235"),
+ Row.of(
+ 12L,
+ "13,0.09134833828228624;10,0.058845768947156235;11,0.058845768947156235"),
+ Row.of(13L, "12,0.09134833828228624")));
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.getConfig().enableObjectReuse();
+ env.setParallelism(4);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ DataStream<Row> dataStream = env.fromCollection(trainRows, trainDataTypeInfo);
+ trainData = tEnv.fromDataStream(dataStream);
+ }
+
+ private void compareResultAndExpected(List<Row> results) {
+ results.sort((o1, o2) -> Long.compare(o1.getFieldAs(0), o2.getFieldAs(0)));
+
+ for (int i = 0; i < results.size(); i++) {
+ Row result = results.get(i);
+ String itemRankScore = result.getFieldAs(1);
+ Row expect = expectedScoreRows.get(i);
+ Assert.assertEquals(result.getField(0), expect.getField(0));
Review Comment:
The expected value should be the first parameter according to its parameter names. So as other places.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(
+ (Long) row.getFieldAs(userCol),
+ (Long) row.getFieldAs(itemCol));
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAllItemsMapState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userItemsMap.entrySet().stream()
Review Comment:
Let's simplify `...stream().forEach(...)` to `...forEach(...)`, if `stream()` is not necessary. IntelliJ also gives further suggestions by using `userItemsMap.forEach` to simplify this part. So as `items.entrySet().stream()` below.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(
+ (Long) row.getFieldAs(userCol),
+ (Long) row.getFieldAs(itemCol));
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
Review Comment:
Since we have just used `Types.LONG` and `Types.STRING`, let's continue using them here to make it consistent.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(
+ (Long) row.getFieldAs(userCol),
+ (Long) row.getFieldAs(itemCol));
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAllItemsMapState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userItemsMap.entrySet().stream()
+ .forEach(
+ entry -> {
+ Long user = entry.getKey();
+ Map<Long, String> items = entry.getValue();
+ if (items.size() >= minUserItemInteraction
+ && items.size() <= maxUserItemInteraction) {
+ items.entrySet().stream()
+ .forEach(
+ item ->
+ output.collect(
+ new StreamRecord<>(
+ new Tuple3<>(
+ user,
+ item.getKey(),
+ items))));
+ }
+ });
+
+ userAllItemsMapState.clear();
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<Long, Long>> element) {
+ Tuple2<Long, Long> userAndItem = element.getValue();
+ long user = userAndItem.f0;
+ long item = userAndItem.f1;
+ Map<Long, String> items = userItemsMap.get(user);
+
+ if (items == null) {
+ items = new LinkedHashMap();
+ }
+
+ if (items.size() <= maxUserItemInteraction) {
+ items.put(item, null);
+ }
+
+ userItemsMap.put(user, items);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ userAllItemsMapState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "userAllItemsMapState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(userAllItemsMapState, "userAllItemsMapState")
+ .ifPresent(
+ stat -> {
+ userItemsMap = stat;
+ });
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ userAllItemsMapState.update(Collections.singletonList(userItemsMap));
+ }
+ }
+
+ /** Calculates top N similar items of each item. */
+ private static class ComputingSimilarItems extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Tuple3<Long, Long, Map<Long, String>>, Row>,
+ BoundedOneInput {
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+ private Map<Long, Map<Long, String>> itemUsersMap = new HashMap<>();
+ private ListState<Map<Long, Map<Long, String>>> userLocalItemsMapState;
+ private ListState<Map<Long, Map<Long, String>>> itemUsersMapState;
+
+ private final int k;
+ private final int maxUserNumPerItem;
+ private final int alpha1;
+ private final int alpha2;
+ private final double beta;
+
+ private static final Character commaDelimiter = ',';
+ private static final Character semicolonDelimiter = ';';
+
+ private ComputingSimilarItems(
+ int k, int maxUserNumPerItem, int alpha1, int alpha2, double beta) {
+ this.k = k;
+ this.maxUserNumPerItem = maxUserNumPerItem;
+ this.alpha1 = alpha1;
+ this.alpha2 = alpha2;
+ this.beta = beta;
+ }
+
+ @Override
+ public void endInput() throws Exception {
+
+ Map<Long, Double> userWeights = new HashMap<>(userItemsMap.size());
+ userItemsMap.forEach(
+ (k, v) -> {
+ int count = v.size();
+ userWeights.put(k, calculateWeight(count));
+ });
+
+ for (long mainItem : itemUsersMap.keySet()) {
+ List<Long> userList = sampleUserList(itemUsersMap.get(mainItem), maxUserNumPerItem);
+ HashMap<Long, Double> id2swing = new HashMap<>();
+
+ for (int i = 0; i < userList.size(); i++) {
+ long u = userList.get(i);
+ for (int j = i + 1; j < userList.size(); j++) {
+ long v = userList.get(j);
+ HashSet<Long> interaction =
Review Comment:
Let's use `Set<Long> interaction = new HashSet<>(userItemsMap.get(u).keySet());` It provides same function with cleaner code.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(
+ (Long) row.getFieldAs(userCol),
+ (Long) row.getFieldAs(itemCol));
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAllItemsMapState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userItemsMap.entrySet().stream()
+ .forEach(
+ entry -> {
+ Long user = entry.getKey();
+ Map<Long, String> items = entry.getValue();
+ if (items.size() >= minUserItemInteraction
+ && items.size() <= maxUserItemInteraction) {
+ items.entrySet().stream()
+ .forEach(
+ item ->
+ output.collect(
+ new StreamRecord<>(
+ new Tuple3<>(
+ user,
+ item.getKey(),
+ items))));
+ }
+ });
+
+ userAllItemsMapState.clear();
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<Long, Long>> element) {
+ Tuple2<Long, Long> userAndItem = element.getValue();
+ long user = userAndItem.f0;
+ long item = userAndItem.f1;
+ Map<Long, String> items = userItemsMap.get(user);
+
+ if (items == null) {
+ items = new LinkedHashMap();
+ }
+
+ if (items.size() <= maxUserItemInteraction) {
+ items.put(item, null);
+ }
+
+ userItemsMap.put(user, items);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ userAllItemsMapState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "userAllItemsMapState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(userAllItemsMapState, "userAllItemsMapState")
+ .ifPresent(
+ stat -> {
+ userItemsMap = stat;
+ });
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ userAllItemsMapState.update(Collections.singletonList(userItemsMap));
+ }
+ }
+
+ /** Calculates top N similar items of each item. */
+ private static class ComputingSimilarItems extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Tuple3<Long, Long, Map<Long, String>>, Row>,
+ BoundedOneInput {
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
Review Comment:
The same issue as `userItemsMap` in `CollectingUserBehavior`.
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/recommendation/SwingTest.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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.recommendation;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.recommendation.swing.Swing;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/** Tests {@link Swing}. */
+public class SwingTest {
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+ private Table trainData;
+ RowTypeInfo trainDataTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
+ },
+ new String[] {"user_id", "item_id"});
+ private static final List<Row> trainRows =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(0L, 10L),
+ Row.of(0L, 11L),
+ Row.of(0L, 12L),
+ Row.of(1L, 13L),
+ Row.of(1L, 12L),
+ Row.of(2L, 10L),
+ Row.of(2L, 11L),
+ Row.of(2L, 12L),
+ Row.of(3L, 13L),
+ Row.of(3L, 12L)));
+
+ private static final List<Row> expectedScoreRows =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(10L, "11,0.058845768947156235;12,0.058845768947156235"),
+ Row.of(11L, "10,0.058845768947156235;12,0.058845768947156235"),
+ Row.of(
+ 12L,
+ "13,0.09134833828228624;10,0.058845768947156235;11,0.058845768947156235"),
+ Row.of(13L, "12,0.09134833828228624")));
+
+ @Before
+ public void before() {
+ env = TestUtils.getExecutionEnvironment();
+ tEnv = StreamTableEnvironment.create(env);
+ trainData = tEnv.fromDataStream(env.fromCollection(trainRows, trainDataTypeInfo));
+ }
+
+ private void compareResultAndExpected(List<Row> results) {
+ results.sort((o1, o2) -> Long.compare(o1.getFieldAs(0), o2.getFieldAs(0)));
+
+ for (int i = 0; i < results.size(); i++) {
+ Row result = results.get(i);
+ String itemRankScore = result.getFieldAs(1);
+ Row expect = expectedScoreRows.get(i);
+ Assert.assertEquals(result.getField(0), expect.getField(0));
+ Assert.assertEquals(itemRankScore, expect.getField(1));
+ }
+ }
+
+ @Test
+ public void testParam() {
+ Swing swing = new Swing();
+
+ assertEquals("item", swing.getItemCol());
+ assertEquals("user", swing.getUserCol());
+ assertEquals(100, swing.getK());
+ assertEquals(1000, swing.getMaxUserNumPerItem());
+ assertEquals(10, swing.getMinUserBehavior());
+ assertEquals(1000, swing.getMaxUserBehavior());
+ assertEquals(15, swing.getAlpha1());
+ assertEquals(0, swing.getAlpha2());
+ assertEquals(0.3, swing.getBeta(), 1e-9);
+
+ swing.setItemCol("item_1")
+ .setUserCol("user_1")
+ .setK(20)
+ .setMaxUserNumPerItem(500)
+ .setMinUserBehavior(10)
+ .setMaxUserBehavior(50)
+ .setAlpha1(5)
+ .setAlpha2(1)
+ .setBeta(0.35);
+
+ assertEquals("item_1", swing.getItemCol());
+ assertEquals("user_1", swing.getUserCol());
+ assertEquals(20, swing.getK());
+ assertEquals(500, swing.getMaxUserNumPerItem());
+ assertEquals(10, swing.getMinUserBehavior());
+ assertEquals(50, swing.getMaxUserBehavior());
+ assertEquals(5, swing.getAlpha1());
+ assertEquals(1, swing.getAlpha2());
+ assertEquals(0.35, swing.getBeta(), 1e-9);
+ }
+
+ @Test
+ public void testDataType() {
+ String msg = "";
+ List<Row> rows =
+ new ArrayList<>(Arrays.asList(Row.of(0, "10"), Row.of(1, "11"), Row.of(2, "")));
+
+ DataStream<Row> dataStream =
+ env.fromCollection(
+ rows,
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {"user_id", "item_id"}));
+ Table data = tEnv.fromDataStream(dataStream);
+
+ try {
+ Table[] swingResultTables =
+ new Swing()
+ .setItemCol("item_id")
+ .setUserCol("user_id")
+ .setOutputCol("item_score")
+ .setMinUserBehavior(1)
+ .transform(data);
+ swingResultTables[0].execute().print();
+ fail();
+ } catch (RuntimeException e) {
+ msg = e.getMessage();
+ assertEquals("The types of user and item columns must be Long.", msg);
+ }
+ }
+
+ @Test
+ public void testNumberFormat() {
+ String msg = "";
Review Comment:
Let's move this declaration to `catch` block, i.e., `String msg = e.getCause()...`. Same as other places.
##########
flink-ml-python/pyflink/ml/recommendation/swing.py:
##########
@@ -0,0 +1,209 @@
+################################################################################
+# 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.
+################################################################################
+import typing
+
+from pyflink.ml.common.param import HasOutputCol
+from pyflink.ml.param import Param, StringParam, IntParam, FloatParam, ParamValidators
+from pyflink.ml.recommendation.common import JavaRecommendationTransformer
+from pyflink.ml.wrapper import JavaWithParams
+
+
+class _SwingParams(
+ JavaWithParams,
+ HasOutputCol
+):
+ """
+ Params for :class:`Swing`.
+ """
+
+ USER_COL: Param[str] = StringParam(
+ "user_col",
+ "Name of user column.",
+ "user",
+ ParamValidators.not_null())
+
+ ITEM_COL: Param[str] = StringParam(
+ "item_col",
+ "Name of item column.",
+ "item",
+ ParamValidators.not_null())
+
+ K: Param[int] = IntParam(
+ "k",
+ "The max number of similar items to output for each item.",
+ 100,
+ ParamValidators.gt(0))
+
+ MAX_USER_NUM_PER_ITEM: Param[int] = IntParam(
+ "max_user_num_per_item",
+ "The max number of users that has purchased for each item. If the number of users that have "
+ + "purchased this item is larger than this value, then only maxUserNumPerItem users will "
+ + "be sampled and used in the computation logic.",
+ 1000,
+ ParamValidators.gt(0))
+
+ MIN_USER_BEHAVIOR: Param[int] = IntParam(
+ "min_user_behavior",
+ "The min number of interaction behavior between item and user.",
+ 10,
+ ParamValidators.gt(0))
+
+ MAX_USER_BEHAVIOR: Param[int] = IntParam(
+ "max_user_behavior",
+ "The max number of interaction behavior between item and user."
+ + "The algorithm filters out activate users.",
+ 1000,
+ ParamValidators.gt(0))
+
+ ALPHA1: Param[int] = IntParam(
+ "alpha2",
+ "This parameter is used to calculate weight of each user.",
+ 15,
+ ParamValidators.gt_eq(0))
+
+ ALPHA2: Param[int] = IntParam(
+ "alpha2",
+ "This parameter is used to calculate similarity of users.",
+ 0,
+ ParamValidators.gt_eq(0))
+
+ BETA: Param[float] = FloatParam(
+ "beta",
+ "This parameter is used to calculate weight of each user.",
+ 0.3,
+ ParamValidators.gt_eq(0))
+
+ def __init__(self, java_params):
+ super(_SwingParams, self).__init__(java_params)
+
+ def set_user_col(self, value: str):
+ return typing.cast(_SwingParams, self.set(self.USER_COL, value))
+
+ def get_user_col(self) -> str:
+ return self.get(self.USER_COL)
+
+ def set_item_col(self, value: str):
+ return typing.cast(_SwingParams, self.set(self.ITEM_COL, value))
+
+ def get_item_col(self) -> str:
+ return self.get(self.ITEM_COL)
+
+ def set_k(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.K, value))
+
+ def get_k(self) -> int:
+ return self.get(self.K)
+
+ def set_max_user_num_per_item(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MAX_USER_NUM_PER_ITEM, value))
+
+ def get_max_user_num_per_item(self) -> int:
+ return self.get(self.MAX_USER_NUM_PER_ITEM)
+
+ def set_min_user_behavior(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MIN_USER_BEHAVIOR, value))
+
+ def get_min_user_behavior(self) -> int:
+ return self.get(self.MIN_USER_BEHAVIOR)
+
+ def set_max_user_behavior(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MAX_USER_BEHAVIOR, value))
+
+ def get_max_user_behavior(self) -> int:
+ return self.get(self.MAX_USER_BEHAVIOR)
+
+ def set_alpha1(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.ALPHA1, value))
+
+ def get_alpha1(self) -> int:
+ return self.get(self.ALPHA1)
+
+ def set_alpha2(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.ALPHA2, value))
+
+ def get_alpha2(self) -> int:
+ return self.get(self.ALPHA2)
+
+ def set_beta(self, value: float):
+ return typing.cast(_SwingParams, self.set(self.BETA, value))
+
+ def get_beta(self) -> float:
+ return self.get(self.BETA)
+
+ @property
+ def user_col(self) -> str:
+ return self.get_user_col()
+
+ @property
+ def item_col(self) -> str:
+ return self.get_item_col()
+
+ @property
+ def k(self) -> int:
+ return self.get_k()
+
+ @property
+ def max_user_num_per_item(self) -> int:
+ return self.get_max_user_num_per_item()
+
+ @property
+ def min_user_behavior(self) -> int:
+ return self.get_min_user_behavior()
+
+ @property
+ def max_user_behavior(self) -> int:
+ return self.get_max_user_behavior()
+
+ @property
+ def alpha1(self) -> int:
+ return self.get_alpha1()
+
+ @property
+ def alpha2(self) -> float:
+ return self.get_alpha2()
+
+ @property
+ def beta(self) -> float:
+ return self.get_beta()
+
+
+class Swing(JavaRecommendationTransformer, _SwingParams):
Review Comment:
In Java code, `Swing` is an implementation of `AlgoOperator`. So here `Swing ` should extends corresponding Python class.
##########
flink-ml-python/pyflink/ml/recommendation/tests/test_swing.py:
##########
@@ -0,0 +1,149 @@
+################################################################################
+# 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.
+################################################################################
+from pyflink.common import Types
+from pyflink.table import Table
+from typing import List
+from py4j.protocol import Py4JJavaError:
+
+from pyflink.ml.recommendation.swing import Swing
+from pyflink.ml.tests.test_utils import PyFlinkMLTestCase
+
+
+# Tests Swing.
+class SwingTest(PyFlinkMLTestCase):
+ def setUp(self):
+ super(SwingTest, self).setUp()
+ self.train_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (0, 11),
+ (0, 12),
+ (1, 13),
+ (1, 12),
+ (2, 10),
+ (2, 11),
+ (2, 12),
+ (3, 13),
+ (3, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.LONG(), Types.LONG()])
+ ))
+
+ self.wrong_type_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (1, 11),
+ (2, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.INT(), Types.LONG()])
+ ))
+
+ self.none_value_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (None, 11),
+ (2, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.LONG(), Types.LONG()])
+ ))
+
+ self.expected_data = [
+ [10, '11,0.058845768947156235;12,0.058845768947156235'],
+ [11, '10,0.058845768947156235;12,0.058845768947156235'],
+ [12, '13,0.09134833828228624;10,0.058845768947156235;11,0.058845768947156235'],
+ [13, '12,0.09134833828228624']]
+
+ def test_param(self):
+ swing = Swing()
+ self.assertEqual("item", swing.item_col)
+ self.assertEqual("user", swing.user_col)
+ self.assertEqual(100, swing.k)
+ self.assertEqual(10, swing.min_user_items)
+ self.assertEqual(1000, swing.max_user_items)
+ self.assertEqual(15, swing.alpha1)
+ self.assertEqual(0, swing.alpha2)
+ self.assertEqual(0.3, swing.beta, delta=1e-9)
+
+ swing.set_item_col("item_1") \
+ .set_user_col("user_1") \
+ .set_k(20) \
+ .set_min_user_behavior(20) \
+ .set_max_user_behavior(50) \
+ .set_alpha1(5) \
+ .set_alpha2(1) \
+ .set_beta(0.35)
+
+ self.assertEqual("item", swing.item_col)
+ self.assertEqual("user", swing.user_col)
+ self.assertEqual(20, swing.k)
+ self.assertEqual(20, swing.min_user_items)
+ self.assertEqual(50, swing.max_user_items)
+ self.assertEqual(5, swing.alpha1)
+ self.assertEqual(1, swing.alpha2)
+ self.assertEqual(0.35, swing.beta, delta=1e-9)
Review Comment:
In Python, we should use `assertAlmostEqual`. The IDE should give warning about this line.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null");
+ }
+ return Tuple2.of(
+ ((Number) row.getFieldAs(userCol)).longValue(),
+ ((Number) row.getFieldAs(itemCol)).longValue());
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAllItemsMapState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userItemsMap.entrySet().stream()
+ .forEach(
+ entry -> {
+ Long user = entry.getKey();
+ Map<Long, String> items = entry.getValue();
+ if (items.size() >= minUserItemInteraction
+ && items.size() <= maxUserItemInteraction) {
+ items.entrySet().stream()
+ .forEach(
+ item ->
+ output.collect(
+ new StreamRecord<>(
+ new Tuple3<>(
+ user,
+ item.getKey(),
+ items))));
+ }
+ });
+
+ userAllItemsMapState.clear();
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<Long, Long>> element) {
+ Tuple2<Long, Long> userAndItem = element.getValue();
+ long user = userAndItem.f0;
+ long item = userAndItem.f1;
+ Map<Long, String> items = userItemsMap.get(user);
+
+ if (items == null) {
+ items = new LinkedHashMap();
+ }
+
+ if (items.size() <= maxUserItemInteraction) {
+ items.put(item, null);
+ }
+
+ userItemsMap.put(user, items);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ userAllItemsMapState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "userAllItemsMapState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(userAllItemsMapState, "userAllItemsMapState")
+ .ifPresent(
+ stat -> {
+ userItemsMap = stat;
+ });
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ userAllItemsMapState.update(Collections.singletonList(userItemsMap));
+ }
+ }
+
+ /** Calculates top N similar items of each item. */
+ private static class ComputingSimilarItems extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Tuple3<Long, Long, Map<Long, String>>, Row>,
+ BoundedOneInput {
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+ private Map<Long, Map<Long, String>> itemUsersMap = new HashMap<>();
+ private ListState<Map<Long, Map<Long, String>>> userLocalItemsMapState;
+ private ListState<Map<Long, Map<Long, String>>> itemUsersMapState;
+
+ private final int k;
+ private final int maxUserNumPerItem;
+ private final int alpha1;
+ private final int alpha2;
+ private final double beta;
+
+ private static Character commaDelimiter = ',';
+ private static Character semicolonDelimiter = ';';
+
+ private ComputingSimilarItems(
+ int k, int maxUserNumPerItem, int alpha1, int alpha2, double beta) {
+ this.k = k;
+ this.maxUserNumPerItem = maxUserNumPerItem;
+ this.alpha1 = alpha1;
+ this.alpha2 = alpha2;
+ this.beta = beta;
+ }
+
+ @Override
+ public void endInput() throws Exception {
+
+ Map<Long, Double> userWeights = new HashMap<>(userItemsMap.size());
+ userItemsMap.forEach(
+ (k, v) -> {
+ int count = v.size();
+ userWeights.put(k, calculateWeight(count));
+ });
+
+ for (long mainItem : itemUsersMap.keySet()) {
+ List<Long> userList = sampleUserList(itemUsersMap.get(mainItem), maxUserNumPerItem);
+ HashMap<Long, Double> id2swing = new HashMap<>();
+
+ for (int i = 0; i < userList.size(); i++) {
+ long u = userList.get(i);
+ for (int j = i + 1; j < userList.size(); j++) {
+ long v = userList.get(j);
+ HashSet<Long> interaction =
+ (HashSet<Long>) new HashSet<>(userItemsMap.get(u).keySet()).clone();
+ interaction.retainAll(userItemsMap.get(v).keySet());
+ if (interaction.size() == 0) {
+ continue;
+ }
+ double similarity =
+ (userWeights.get(u)
+ * userWeights.get(v)
+ / (alpha2 + interaction.size()));
+ for (long simItem : interaction) {
+ if (simItem == mainItem) {
+ continue;
+ }
+ double itemSimilarity =
+ id2swing.getOrDefault(simItem, 0.0) + similarity;
+ id2swing.putIfAbsent(simItem, itemSimilarity);
+ }
+ }
+ }
+
+ ArrayList<Tuple2<Long, Double>> itemAndScore = new ArrayList<>();
+ id2swing.forEach((key, value) -> itemAndScore.add(Tuple2.of(key, value)));
+
+ itemAndScore.sort((o1, o2) -> Double.compare(o2.f1, o1.f1));
+
+ if (itemAndScore.size() == 0) {
+ continue;
+ }
+
+ int itemNums = Math.min(k, itemAndScore.size());
+ StringBuilder sbd = new StringBuilder();
+ for (int i = 0; i < itemNums; i++) {
+ sbd.append(itemAndScore.get(i).f0).append(commaDelimiter);
+ sbd.append(itemAndScore.get(i).f1).append(semicolonDelimiter);
+ }
+ String itemList = sbd.substring(0, sbd.length() - 1);
+
+ output.collect(new StreamRecord<>(Row.of(mainItem, itemList)));
+ }
+
+ userLocalItemsMapState.clear();
+ itemUsersMapState.clear();
+ }
+
+ private double calculateWeight(int size) {
+ return (1.0 / Math.pow(alpha1 + size, beta));
+ }
+
+ private static List<Long> sampleUserList(Map<Long, String> allUsers, int sampleSize) {
+ int totalSize = allUsers.size();
+ if (totalSize < sampleSize) {
+ return new ArrayList(allUsers.keySet());
+ }
+
+ List<Long> userList = new ArrayList<>(totalSize);
+ double prob = (double) sampleSize / totalSize;
+ Random rand = new Random();
Review Comment:
Use `Random` without seed will make the results unreproducible. Is it possible to use deterministic seeds here?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
Review Comment:
It is vague to say `formula of Swing`. It's better to change `The formula of Swing is` to `The similarity between items in Swing is defined as`. So as in Python document.
##########
flink-ml-python/pyflink/ml/lib/recommendation/swing.py:
##########
@@ -0,0 +1,209 @@
+################################################################################
+# 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.
+################################################################################
+import typing
+
+from pyflink.ml.core.param import Param, StringParam, IntParam, FloatParam, ParamValidators
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.param import HasOutputCol
+from pyflink.ml.lib.recommendation.common import JavaRecommendationTransformer
+
+
+class _SwingParams(
+ JavaWithParams,
+ HasOutputCol
+):
+ """
+ Params for :class:`Swing`.
+ """
+
+ USER_COL: Param[str] = StringParam(
+ "user_col",
+ "Name of user column.",
+ "user",
+ ParamValidators.not_null())
+
+ ITEM_COL: Param[str] = StringParam(
+ "item_col",
+ "Name of item column.",
+ "item",
+ ParamValidators.not_null())
+
+ K: Param[int] = IntParam(
+ "k",
+ "The max number of similar items to output for each item.",
+ 100,
+ ParamValidators.gt(0))
+
+ MAX_USER_NUM_PER_ITEM: Param[int] = IntParam(
+ "max_user_num_per_item",
+ "The max number of users that has purchased for each item. If the number of users that have "
+ + "purchased this item is larger than this value, then only maxUserNumPerItem users will "
+ + "be sampled and used in the computation logic.",
+ 1000,
+ ParamValidators.gt(0))
+
+ MIN_USER_BEHAVIOR: Param[int] = IntParam(
+ "min_user_behavior",
+ "The min number of interaction behavior between item and user.",
+ 10,
+ ParamValidators.gt(0))
+
+ MAX_USER_BEHAVIOR: Param[int] = IntParam(
+ "max_user_behavior",
+ "The max number of interaction behavior between item and user."
+ + "The algorithm filters out activate users.",
+ 1000,
+ ParamValidators.gt(0))
+
+ ALPHA1: Param[int] = IntParam(
+ "alpha2",
+ "This parameter is used to calculate weight of each user.",
+ 15,
+ ParamValidators.gt_eq(0))
+
+ ALPHA2: Param[int] = IntParam(
+ "alpha2",
+ "This parameter is used to calculate similarity of users.",
+ 0,
+ ParamValidators.gt_eq(0))
+
+ BETA: Param[float] = FloatParam(
+ "beta",
+ "This parameter is used to calculate weight of each user.",
+ 0.3,
+ ParamValidators.gt_eq(0))
+
+ def __init__(self, java_params):
+ super(_SwingParams, self).__init__(java_params)
+
+ def set_user_col(self, value: str):
+ return typing.cast(_SwingParams, self.set(self.USER_COL, value))
+
+ def get_user_col(self) -> str:
+ return self.get(self.USER_COL)
+
+ def set_item_col(self, value: str):
+ return typing.cast(_SwingParams, self.set(self.ITEM_COL, value))
+
+ def get_item_col(self) -> str:
+ return self.get(self.ITEM_COL)
+
+ def set_k(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.K, value))
+
+ def get_k(self) -> int:
+ return self.get(self.K)
+
+ def set_max_user_num_per_item(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MAX_USER_NUM_PER_ITEM, value))
+
+ def get_max_user_num_per_item(self) -> int:
+ return self.get(self.MAX_USER_NUM_PER_ITEM)
+
+ def set_min_user_behavior(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MIN_USER_BEHAVIOR, value))
+
+ def get_min_user_behavior(self) -> int:
+ return self.get(self.MIN_USER_BEHAVIOR)
+
+ def set_max_user_behavior(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.MAX_USER_BEHAVIOR, value))
+
+ def get_max_user_behavior(self) -> int:
+ return self.get(self.MAX_USER_BEHAVIOR)
+
+ def set_alpha1(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.ALPHA1, value))
+
+ def get_alpha1(self) -> int:
+ return self.get(self.ALPHA1)
+
+ def set_alpha2(self, value: int):
+ return typing.cast(_SwingParams, self.set(self.ALPHA2, value))
+
+ def get_alpha2(self) -> int:
+ return self.get(self.ALPHA2)
+
+ def set_beta(self, value: float):
+ return typing.cast(_SwingParams, self.set(self.BETA, value))
+
+ def get_beta(self) -> float:
+ return self.get(self.BETA)
+
+ @property
+ def user_col(self) -> str:
+ return self.get_user_col()
+
+ @property
+ def item_col(self) -> str:
+ return self.get_item_col()
+
+ @property
+ def k(self) -> int:
+ return self.get_k()
+
+ @property
+ def max_user_num_per_item(self) -> int:
+ return self.get_max_user_num_per_item()
+
+ @property
+ def min_user_behavior(self) -> int:
+ return self.get_min_user_behavior()
+
+ @property
+ def max_user_behavior(self) -> int:
+ return self.get_max_user_behavior()
+
+ @property
+ def alpha1(self) -> int:
+ return self.get_alpha1()
+
+ @property
+ def alpha2(self) -> float:
+ return self.get_alpha2()
+
+ @property
+ def beta(self) -> float:
+ return self.get_beta()
+
+
+class Swing(JavaRecommendationTransformer, _SwingParams):
+ """
+ An AlgoOperator which implements the Swing algorithm.
+
+ Swing is an item recall model. The topology of user-item graph usually can be described as
+ user-item-user or item-user-item, which are like 'swing'. For example, if both user u
+ and user v have purchased the same commodity i , they will form a relationship
+ diagram similar to a swing. If u and v have purchased commodity j in
+ addition to i, it is supposed i and j are similar. The formula of Swing is
+ $$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap I_v|}} $$
Review Comment:
`\alpha\_2` should be `\alpha_2`, so as in Java document.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
Review Comment:
Let's have a check between `minUserBehavior` and `maxUserBehavior` here, in case `minUserBehavior` is larger than `maxUserBehavior`.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(
+ (Long) row.getFieldAs(userCol),
+ (Long) row.getFieldAs(itemCol));
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAllItemsMapState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userItemsMap.entrySet().stream()
+ .forEach(
+ entry -> {
+ Long user = entry.getKey();
+ Map<Long, String> items = entry.getValue();
+ if (items.size() >= minUserItemInteraction
+ && items.size() <= maxUserItemInteraction) {
+ items.entrySet().stream()
+ .forEach(
+ item ->
+ output.collect(
+ new StreamRecord<>(
+ new Tuple3<>(
+ user,
+ item.getKey(),
+ items))));
+ }
+ });
+
+ userAllItemsMapState.clear();
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<Long, Long>> element) {
+ Tuple2<Long, Long> userAndItem = element.getValue();
+ long user = userAndItem.f0;
+ long item = userAndItem.f1;
+ Map<Long, String> items = userItemsMap.get(user);
+
+ if (items == null) {
+ items = new LinkedHashMap();
Review Comment:
```suggestion
items = new LinkedHashMap<>();
```
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(
+ (Long) row.getFieldAs(userCol),
+ (Long) row.getFieldAs(itemCol));
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
Review Comment:
I'm confused about its type. In `endInput`, values of inner maps are never used. In `processElement`, the values of inner maps are always `null` (set by `items.put(item, null);`). Then, can we change its type to `Map<Long, Set<Long>>`?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(
+ (Long) row.getFieldAs(userCol),
+ (Long) row.getFieldAs(itemCol));
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAllItemsMapState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userItemsMap.entrySet().stream()
+ .forEach(
+ entry -> {
+ Long user = entry.getKey();
+ Map<Long, String> items = entry.getValue();
+ if (items.size() >= minUserItemInteraction
+ && items.size() <= maxUserItemInteraction) {
+ items.entrySet().stream()
+ .forEach(
+ item ->
+ output.collect(
+ new StreamRecord<>(
+ new Tuple3<>(
+ user,
+ item.getKey(),
+ items))));
+ }
+ });
+
+ userAllItemsMapState.clear();
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<Long, Long>> element) {
+ Tuple2<Long, Long> userAndItem = element.getValue();
+ long user = userAndItem.f0;
+ long item = userAndItem.f1;
+ Map<Long, String> items = userItemsMap.get(user);
+
+ if (items == null) {
+ items = new LinkedHashMap();
+ }
+
+ if (items.size() <= maxUserItemInteraction) {
+ items.put(item, null);
+ }
+
+ userItemsMap.put(user, items);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ userAllItemsMapState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "userAllItemsMapState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(userAllItemsMapState, "userAllItemsMapState")
+ .ifPresent(
+ stat -> {
+ userItemsMap = stat;
+ });
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ userAllItemsMapState.update(Collections.singletonList(userItemsMap));
+ }
+ }
+
+ /** Calculates top N similar items of each item. */
+ private static class ComputingSimilarItems extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Tuple3<Long, Long, Map<Long, String>>, Row>,
+ BoundedOneInput {
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+ private Map<Long, Map<Long, String>> itemUsersMap = new HashMap<>();
+ private ListState<Map<Long, Map<Long, String>>> userLocalItemsMapState;
+ private ListState<Map<Long, Map<Long, String>>> itemUsersMapState;
+
+ private final int k;
+ private final int maxUserNumPerItem;
+ private final int alpha1;
+ private final int alpha2;
+ private final double beta;
+
+ private static final Character commaDelimiter = ',';
+ private static final Character semicolonDelimiter = ';';
+
+ private ComputingSimilarItems(
+ int k, int maxUserNumPerItem, int alpha1, int alpha2, double beta) {
+ this.k = k;
+ this.maxUserNumPerItem = maxUserNumPerItem;
+ this.alpha1 = alpha1;
+ this.alpha2 = alpha2;
+ this.beta = beta;
+ }
+
+ @Override
+ public void endInput() throws Exception {
+
+ Map<Long, Double> userWeights = new HashMap<>(userItemsMap.size());
+ userItemsMap.forEach(
+ (k, v) -> {
+ int count = v.size();
+ userWeights.put(k, calculateWeight(count));
+ });
+
+ for (long mainItem : itemUsersMap.keySet()) {
+ List<Long> userList = sampleUserList(itemUsersMap.get(mainItem), maxUserNumPerItem);
+ HashMap<Long, Double> id2swing = new HashMap<>();
+
+ for (int i = 0; i < userList.size(); i++) {
+ long u = userList.get(i);
+ for (int j = i + 1; j < userList.size(); j++) {
+ long v = userList.get(j);
+ HashSet<Long> interaction =
+ (HashSet<Long>) new HashSet<>(userItemsMap.get(u).keySet()).clone();
+ interaction.retainAll(userItemsMap.get(v).keySet());
+ if (interaction.size() == 0) {
+ continue;
+ }
+ double similarity =
+ (userWeights.get(u)
+ * userWeights.get(v)
+ / (alpha2 + interaction.size()));
+ for (long simItem : interaction) {
+ if (simItem == mainItem) {
+ continue;
+ }
+ double itemSimilarity =
+ id2swing.getOrDefault(simItem, 0.0) + similarity;
+ id2swing.putIfAbsent(simItem, itemSimilarity);
+ }
+ }
+ }
+
+ ArrayList<Tuple2<Long, Double>> itemAndScore = new ArrayList<>();
+ id2swing.forEach((key, value) -> itemAndScore.add(Tuple2.of(key, value)));
+
+ itemAndScore.sort((o1, o2) -> Double.compare(o2.f1, o1.f1));
+
+ if (itemAndScore.size() == 0) {
+ continue;
+ }
+
+ int itemNums = Math.min(k, itemAndScore.size());
+ StringBuilder sbd = new StringBuilder();
+ for (int i = 0; i < itemNums; i++) {
+ sbd.append(itemAndScore.get(i).f0).append(commaDelimiter);
+ sbd.append(itemAndScore.get(i).f1).append(semicolonDelimiter);
+ }
+ String itemList = sbd.substring(0, sbd.length() - 1);
+
+ output.collect(new StreamRecord<>(Row.of(mainItem, itemList)));
+ }
+
+ userLocalItemsMapState.clear();
+ itemUsersMapState.clear();
+ }
+
+ private double calculateWeight(int size) {
+ return (1.0 / Math.pow(alpha1 + size, beta));
+ }
+
+ private static List<Long> sampleUserList(Map<Long, String> allUsers, int sampleSize) {
+ int totalSize = allUsers.size();
+ if (totalSize < sampleSize) {
+ return new ArrayList(allUsers.keySet());
+ }
+
+ List<Long> userList = new ArrayList<>(totalSize);
+ double prob = (double) sampleSize / totalSize;
+ Random rand = new Random();
+
+ for (long u : allUsers.keySet()) {
Review Comment:
Current sampling may not accurately return `sampleSize` elements.
It seems not necessary to use streaming sampling here. Can we use `Collections.shuffle` then return first `sampleSize ` elements?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java:
##########
@@ -0,0 +1,430 @@
+/*
+ * 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.recommendation.swing;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+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.typeutils.RowTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.table.catalog.ResolvedSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall model. The topology of user-item graph usually can be described as
+ * user-item-user or item-user-item, which are like 'swing'. For example, if both user <em>u</em>
+ * and user <em>v</em> have purchased the same commodity <em>i</em> , they will form a relationship
+ * diagram similar to a swing. If <em>u</em> and <em>v</em> have purchased commodity <em>j</em> in
+ * addition to <em>i</em>, it is supposed <em>i</em> and <em>j</em> are similar. The formula of
+ * Swing is
+ *
+ * <p>$$ w_{(i,j)}=\sum_{u\in U_i\cap U_j}\sum_{v\in U_i\cap
+ * U_j}{\frac{1}{{(I_u+\alpha_1)}^\beta}}*{\frac{1}{{(I_v+\alpha_1)}^\beta}}*{\frac{1}{\alpha\_2+|I_u\cap
+ * I_v|}} $$
+ *
+ * <p>This implementation is based on the algorithm proposed in the paper: "Large Scale Product
+ * Graph Construction for Recommendation in E-commerce" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ * (https://arxiv.org/pdf/2010.05525.pdf)
+ */
+public class Swing implements AlgoOperator<Swing>, SwingParams<Swing> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public Swing() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ Preconditions.checkArgument(inputs.length == 1);
+ final ResolvedSchema schema = inputs[0].getResolvedSchema();
+
+ if (!(Types.LONG.equals(TableUtils.getTypeInfoByName(schema, userCol))
+ && Types.LONG.equals(TableUtils.getTypeInfoByName(schema, itemCol)))) {
+ throw new IllegalArgumentException("The types of user and item columns must be Long.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> itemUsers =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ if (row.getFieldAs(userCol) == null
+ || row.getFieldAs(itemCol) == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(
+ (Long) row.getFieldAs(userCol),
+ (Long) row.getFieldAs(itemCol));
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, Map<Long, String>>> userAllItemsStream =
+ itemUsers
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "fillUserItemsTable",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING)),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userAllItemsStream
+ .keyBy(tuple -> tuple.f1)
+ .transform(
+ "computingSimilarItems",
+ outputTypeInfo,
+ new ComputingSimilarItems(
+ getK(),
+ getMaxUserNumPerItem(),
+ getAlpha1(),
+ getAlpha2(),
+ getBeta()));
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static Swing load(StreamTableEnvironment tEnv, String path) throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ /**
+ * Appends one column, that records all items the user has purchased, to the input table.
+ *
+ * <p>During the process, this operator collect users and all items a user has purchased into a
+ * map of list. When the input is finished, this operator appends the certain
+ * user-purchased-items list to each row.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, Map<Long, String>>>
+ implements OneInputStreamOperator<
+ Tuple2<Long, Long>, Tuple3<Long, Long, Map<Long, String>>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAllItemsMapState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userItemsMap.entrySet().stream()
+ .forEach(
+ entry -> {
+ Long user = entry.getKey();
+ Map<Long, String> items = entry.getValue();
+ if (items.size() >= minUserItemInteraction
+ && items.size() <= maxUserItemInteraction) {
+ items.entrySet().stream()
+ .forEach(
+ item ->
+ output.collect(
+ new StreamRecord<>(
+ new Tuple3<>(
+ user,
+ item.getKey(),
+ items))));
+ }
+ });
+
+ userAllItemsMapState.clear();
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple2<Long, Long>> element) {
+ Tuple2<Long, Long> userAndItem = element.getValue();
+ long user = userAndItem.f0;
+ long item = userAndItem.f1;
+ Map<Long, String> items = userItemsMap.get(user);
+
+ if (items == null) {
+ items = new LinkedHashMap();
+ }
+
+ if (items.size() <= maxUserItemInteraction) {
+ items.put(item, null);
+ }
+
+ userItemsMap.put(user, items);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ userAllItemsMapState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "userAllItemsMapState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(userAllItemsMapState, "userAllItemsMapState")
+ .ifPresent(
+ stat -> {
+ userItemsMap = stat;
+ });
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ userAllItemsMapState.update(Collections.singletonList(userItemsMap));
+ }
+ }
+
+ /** Calculates top N similar items of each item. */
+ private static class ComputingSimilarItems extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Tuple3<Long, Long, Map<Long, String>>, Row>,
+ BoundedOneInput {
+
+ private Map<Long, Map<Long, String>> userItemsMap = new HashMap<>();
+ private Map<Long, Map<Long, String>> itemUsersMap = new HashMap<>();
+ private ListState<Map<Long, Map<Long, String>>> userLocalItemsMapState;
+ private ListState<Map<Long, Map<Long, String>>> itemUsersMapState;
+
+ private final int k;
+ private final int maxUserNumPerItem;
+ private final int alpha1;
+ private final int alpha2;
+ private final double beta;
+
+ private static final Character commaDelimiter = ',';
+ private static final Character semicolonDelimiter = ';';
+
+ private ComputingSimilarItems(
+ int k, int maxUserNumPerItem, int alpha1, int alpha2, double beta) {
+ this.k = k;
+ this.maxUserNumPerItem = maxUserNumPerItem;
+ this.alpha1 = alpha1;
+ this.alpha2 = alpha2;
+ this.beta = beta;
+ }
+
+ @Override
+ public void endInput() throws Exception {
+
+ Map<Long, Double> userWeights = new HashMap<>(userItemsMap.size());
+ userItemsMap.forEach(
+ (k, v) -> {
+ int count = v.size();
+ userWeights.put(k, calculateWeight(count));
+ });
+
+ for (long mainItem : itemUsersMap.keySet()) {
+ List<Long> userList = sampleUserList(itemUsersMap.get(mainItem), maxUserNumPerItem);
+ HashMap<Long, Double> id2swing = new HashMap<>();
+
+ for (int i = 0; i < userList.size(); i++) {
+ long u = userList.get(i);
+ for (int j = i + 1; j < userList.size(); j++) {
+ long v = userList.get(j);
+ HashSet<Long> interaction =
+ (HashSet<Long>) new HashSet<>(userItemsMap.get(u).keySet()).clone();
+ interaction.retainAll(userItemsMap.get(v).keySet());
+ if (interaction.size() == 0) {
+ continue;
+ }
+ double similarity =
+ (userWeights.get(u)
+ * userWeights.get(v)
+ / (alpha2 + interaction.size()));
+ for (long simItem : interaction) {
+ if (simItem == mainItem) {
+ continue;
+ }
+ double itemSimilarity =
+ id2swing.getOrDefault(simItem, 0.0) + similarity;
+ id2swing.putIfAbsent(simItem, itemSimilarity);
+ }
+ }
+ }
+
+ ArrayList<Tuple2<Long, Double>> itemAndScore = new ArrayList<>();
+ id2swing.forEach((key, value) -> itemAndScore.add(Tuple2.of(key, value)));
+
+ itemAndScore.sort((o1, o2) -> Double.compare(o2.f1, o1.f1));
+
+ if (itemAndScore.size() == 0) {
+ continue;
+ }
+
+ int itemNums = Math.min(k, itemAndScore.size());
+ StringBuilder sbd = new StringBuilder();
+ for (int i = 0; i < itemNums; i++) {
+ sbd.append(itemAndScore.get(i).f0).append(commaDelimiter);
+ sbd.append(itemAndScore.get(i).f1).append(semicolonDelimiter);
+ }
+ String itemList = sbd.substring(0, sbd.length() - 1);
+
+ output.collect(new StreamRecord<>(Row.of(mainItem, itemList)));
+ }
+
+ userLocalItemsMapState.clear();
+ itemUsersMapState.clear();
+ }
+
+ private double calculateWeight(int size) {
+ return (1.0 / Math.pow(alpha1 + size, beta));
+ }
+
+ private static List<Long> sampleUserList(Map<Long, String> allUsers, int sampleSize) {
+ int totalSize = allUsers.size();
+ if (totalSize < sampleSize) {
+ return new ArrayList(allUsers.keySet());
Review Comment:
```suggestion
return new ArrayList<>(allUsers.keySet());
```
##########
flink-ml-python/pyflink/examples/ml/recommendation/swing_example.py:
##########
@@ -0,0 +1,72 @@
+################################################################################
+# 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.
+################################################################################
+
+# Simple program that creates a Swing instance and gives recommendations for items.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.table import StreamTableEnvironment
+
+from pyflink.ml.recommendation.swing import Swing
+
+# Creates a new StreamExecutionEnvironment.
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# Creates a StreamTableEnvironment.
+t_env = StreamTableEnvironment.create(env)
+
+# Generates input data.
+input_table = t_env.from_data_stream(
+ env.from_collection([
+ (0, 10),
+ (0, 11),
+ (0, 12),
+ (1, 13),
+ (1, 12),
+ (2, 10),
+ (2, 11),
+ (2, 12),
+ (3, 13),
+ (3, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.LONG(), Types.LONG()])
+ ))
+
+# Creates a swing object and initialize its parameters.
+swing = Swing()\
+ .set_item_col('item')\
+ .set_user_col("user")\
+ .set_min_user_behavior(1)
+
+# Transforms the data to Swing algorithm result.
+swingTable = swing.transform(input_table)
Review Comment:
`swingTable` is ambiguous and also not conform to Python naming convention. Let's use `output` instead like other tests.
##########
flink-ml-python/pyflink/examples/ml/recommendation/swing_example.py:
##########
@@ -0,0 +1,72 @@
+################################################################################
+# 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.
+################################################################################
+
+# Simple program that creates a Swing instance and gives recommendations for items.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.table import StreamTableEnvironment
+
+from pyflink.ml.recommendation.swing import Swing
+
+# Creates a new StreamExecutionEnvironment.
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# Creates a StreamTableEnvironment.
+t_env = StreamTableEnvironment.create(env)
+
+# Generates input data.
+input_table = t_env.from_data_stream(
+ env.from_collection([
+ (0, 10),
+ (0, 11),
+ (0, 12),
+ (1, 13),
+ (1, 12),
+ (2, 10),
+ (2, 11),
+ (2, 12),
+ (3, 13),
+ (3, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.LONG(), Types.LONG()])
+ ))
+
+# Creates a swing object and initialize its parameters.
+swing = Swing()\
+ .set_item_col('item')\
+ .set_user_col("user")\
+ .set_min_user_behavior(1)
+
+# Transforms the data to Swing algorithm result.
+swingTable = swing.transform(input_table)
+
+# Extracts and display the results.
+field_names = swingTable[0].get_schema().get_field_names()
+
+results = t_env.to_changelog_stream(
Review Comment:
Let's `to_data_stream` here like other tests, if `to_changelog_stream` is not necessary.
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/recommendation/SwingTest.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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.recommendation;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.recommendation.swing.Swing;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/** Tests {@link Swing}. */
+public class SwingTest {
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+ private Table trainData;
+ RowTypeInfo trainDataTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
+ },
+ new String[] {"user_id", "item_id"});
+ private static final List<Row> trainRows =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(0L, 10L),
+ Row.of(0L, 11L),
+ Row.of(0L, 12L),
+ Row.of(1L, 13L),
+ Row.of(1L, 12L),
+ Row.of(2L, 10L),
+ Row.of(2L, 11L),
+ Row.of(2L, 12L),
+ Row.of(3L, 13L),
+ Row.of(3L, 12L)));
+
+ private static final List<Row> expectedScoreRows =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(10L, "11,0.058845768947156235;12,0.058845768947156235"),
+ Row.of(11L, "10,0.058845768947156235;12,0.058845768947156235"),
+ Row.of(
+ 12L,
+ "13,0.09134833828228624;10,0.058845768947156235;11,0.058845768947156235"),
+ Row.of(13L, "12,0.09134833828228624")));
+
+ @Before
+ public void before() {
+ env = TestUtils.getExecutionEnvironment();
+ tEnv = StreamTableEnvironment.create(env);
+ trainData = tEnv.fromDataStream(env.fromCollection(trainRows, trainDataTypeInfo));
+ }
+
+ private void compareResultAndExpected(List<Row> results) {
+ results.sort((o1, o2) -> Long.compare(o1.getFieldAs(0), o2.getFieldAs(0)));
+
+ for (int i = 0; i < results.size(); i++) {
+ Row result = results.get(i);
+ String itemRankScore = result.getFieldAs(1);
+ Row expect = expectedScoreRows.get(i);
+ Assert.assertEquals(result.getField(0), expect.getField(0));
+ Assert.assertEquals(itemRankScore, expect.getField(1));
+ }
+ }
+
+ @Test
+ public void testParam() {
+ Swing swing = new Swing();
+
+ assertEquals("item", swing.getItemCol());
+ assertEquals("user", swing.getUserCol());
+ assertEquals(100, swing.getK());
+ assertEquals(1000, swing.getMaxUserNumPerItem());
+ assertEquals(10, swing.getMinUserBehavior());
+ assertEquals(1000, swing.getMaxUserBehavior());
+ assertEquals(15, swing.getAlpha1());
+ assertEquals(0, swing.getAlpha2());
+ assertEquals(0.3, swing.getBeta(), 1e-9);
+
+ swing.setItemCol("item_1")
+ .setUserCol("user_1")
+ .setK(20)
+ .setMaxUserNumPerItem(500)
+ .setMinUserBehavior(10)
+ .setMaxUserBehavior(50)
+ .setAlpha1(5)
+ .setAlpha2(1)
+ .setBeta(0.35);
+
+ assertEquals("item_1", swing.getItemCol());
+ assertEquals("user_1", swing.getUserCol());
+ assertEquals(20, swing.getK());
+ assertEquals(500, swing.getMaxUserNumPerItem());
+ assertEquals(10, swing.getMinUserBehavior());
+ assertEquals(50, swing.getMaxUserBehavior());
+ assertEquals(5, swing.getAlpha1());
+ assertEquals(1, swing.getAlpha2());
+ assertEquals(0.35, swing.getBeta(), 1e-9);
+ }
+
+ @Test
+ public void testDataType() {
+ String msg = "";
+ List<Row> rows =
+ new ArrayList<>(Arrays.asList(Row.of(0, "10"), Row.of(1, "11"), Row.of(2, "")));
+
+ DataStream<Row> dataStream =
+ env.fromCollection(
+ rows,
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ },
+ new String[] {"user_id", "item_id"}));
+ Table data = tEnv.fromDataStream(dataStream);
+
+ try {
+ Table[] swingResultTables =
+ new Swing()
+ .setItemCol("item_id")
+ .setUserCol("user_id")
+ .setOutputCol("item_score")
+ .setMinUserBehavior(1)
+ .transform(data);
+ swingResultTables[0].execute().print();
+ fail();
+ } catch (RuntimeException e) {
+ msg = e.getMessage();
+ assertEquals("The types of user and item columns must be Long.", msg);
+ }
+ }
+
+ @Test
+ public void testNumberFormat() {
+ String msg = "";
+ List<Row> rows =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(0L, 10L),
+ Row.of(null, 12L),
+ Row.of(1L, 13L),
+ Row.of(3L, 12L)));
+
+ DataStream<Row> dataStream = env.fromCollection(rows, trainDataTypeInfo);
+ Table data = tEnv.fromDataStream(dataStream);
+ Swing swing = new Swing().setItemCol("item_id").setUserCol("user_id").setMinUserBehavior(1);
+ Table[] swingResultTables = swing.transform(data);
+
+ try {
+ swingResultTables[0].execute().print();
+ fail();
+ } catch (RuntimeException e) {
+ msg = e.getCause().getCause().getCause().getCause().getCause().getMessage();
Review Comment:
We can replace this line with `ExceptionUtils.getRootCause`. You can check its usage in other tests.
##########
flink-ml-python/pyflink/ml/recommendation/__init__.py:
##########
@@ -0,0 +1,17 @@
+################################################################################
+# 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.
+################################################################################
Review Comment:
There should be a newline here.
There are some other code style problems. You can check your code style and types locally. Please refer to related commands here: [1]
[1] [ - name: Check code style](https://github.com/apache/flink-ml/blob/76d5b864dd7df95adebebe6f7e47d52ae0f689c7/.github/workflows/python-tests.yml#L54)
##########
flink-ml-python/pyflink/ml/recommendation/tests/test_swing.py:
##########
@@ -0,0 +1,149 @@
+################################################################################
+# 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.
+################################################################################
+from pyflink.common import Types
+from pyflink.table import Table
+from typing import List
+from py4j.protocol import Py4JJavaError:
+
+from pyflink.ml.recommendation.swing import Swing
+from pyflink.ml.tests.test_utils import PyFlinkMLTestCase
+
+
+# Tests Swing.
+class SwingTest(PyFlinkMLTestCase):
+ def setUp(self):
+ super(SwingTest, self).setUp()
+ self.train_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (0, 11),
+ (0, 12),
+ (1, 13),
+ (1, 12),
+ (2, 10),
+ (2, 11),
+ (2, 12),
+ (3, 13),
+ (3, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.LONG(), Types.LONG()])
+ ))
+
+ self.wrong_type_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (1, 11),
+ (2, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.INT(), Types.LONG()])
+ ))
+
+ self.none_value_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (None, 11),
+ (2, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.LONG(), Types.LONG()])
+ ))
+
+ self.expected_data = [
+ [10, '11,0.058845768947156235;12,0.058845768947156235'],
+ [11, '10,0.058845768947156235;12,0.058845768947156235'],
+ [12, '13,0.09134833828228624;10,0.058845768947156235;11,0.058845768947156235'],
+ [13, '12,0.09134833828228624']]
+
+ def test_param(self):
+ swing = Swing()
+ self.assertEqual("item", swing.item_col)
+ self.assertEqual("user", swing.user_col)
+ self.assertEqual(100, swing.k)
+ self.assertEqual(10, swing.min_user_items)
+ self.assertEqual(1000, swing.max_user_items)
+ self.assertEqual(15, swing.alpha1)
+ self.assertEqual(0, swing.alpha2)
+ self.assertEqual(0.3, swing.beta, delta=1e-9)
Review Comment:
In Python, we should use `assertAlmostEqual`. The IDE should give warning about this line.
##########
flink-ml-python/pyflink/ml/recommendation/tests/test_swing.py:
##########
@@ -0,0 +1,149 @@
+################################################################################
+# 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.
+################################################################################
+from pyflink.common import Types
+from pyflink.table import Table
+from typing import List
+from py4j.protocol import Py4JJavaError:
Review Comment:
Let's run this test locally to ensure it can compile.
```suggestion
from py4j.protocol import Py4JJavaError
```
##########
flink-ml-python/pyflink/ml/recommendation/tests/test_swing.py:
##########
@@ -0,0 +1,149 @@
+################################################################################
+# 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.
+################################################################################
+from pyflink.common import Types
+from pyflink.table import Table
+from typing import List
+from py4j.protocol import Py4JJavaError:
+
+from pyflink.ml.recommendation.swing import Swing
+from pyflink.ml.tests.test_utils import PyFlinkMLTestCase
+
+
+# Tests Swing.
+class SwingTest(PyFlinkMLTestCase):
+ def setUp(self):
+ super(SwingTest, self).setUp()
+ self.train_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (0, 11),
+ (0, 12),
+ (1, 13),
+ (1, 12),
+ (2, 10),
+ (2, 11),
+ (2, 12),
+ (3, 13),
+ (3, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.LONG(), Types.LONG()])
+ ))
+
+ self.wrong_type_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (1, 11),
+ (2, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.INT(), Types.LONG()])
+ ))
+
+ self.none_value_data = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (0, 10),
+ (None, 11),
+ (2, 12)
+ ],
+ type_info=Types.ROW_NAMED(
+ ['user', 'item'],
+ [Types.LONG(), Types.LONG()])
+ ))
+
+ self.expected_data = [
+ [10, '11,0.058845768947156235;12,0.058845768947156235'],
+ [11, '10,0.058845768947156235;12,0.058845768947156235'],
+ [12, '13,0.09134833828228624;10,0.058845768947156235;11,0.058845768947156235'],
+ [13, '12,0.09134833828228624']]
+
+ def test_param(self):
+ swing = Swing()
+ self.assertEqual("item", swing.item_col)
+ self.assertEqual("user", swing.user_col)
+ self.assertEqual(100, swing.k)
+ self.assertEqual(10, swing.min_user_items)
Review Comment:
There is no `min_user_items` and `max_user_items` defined in `Swing`. Please make sure this test is passed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org