You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2023/02/16 11:27:00 UTC
[flink-ml] branch master updated: [FLINK-30451] Add AlgoOperator for Swing
This is an automated email from the ASF dual-hosted git repository.
zhangzp pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-ml.git
The following commit(s) were added to refs/heads/master by this push:
new e85fcf7 [FLINK-30451] Add AlgoOperator for Swing
e85fcf7 is described below
commit e85fcf7635154871a49f6feb53553b0e98b21d88
Author: vacaly <37...@users.noreply.github.com>
AuthorDate: Thu Feb 16 19:26:54 2023 +0800
[FLINK-30451] Add AlgoOperator for Swing
This closes #192.
---
.../docs/operators/recommendation/_index.md | 25 ++
.../content/docs/operators/recommendation/swing.md | 194 ++++++++++
.../ml/examples/recommendation/SwingExample.java | 69 ++++
.../flink/ml/recommendation/swing/Swing.java | 417 +++++++++++++++++++++
.../flink/ml/recommendation/swing/SwingParams.java | 168 +++++++++
.../apache/flink/ml/recommendation/SwingTest.java | 230 ++++++++++++
.../examples/ml/recommendation/swing_example.py | 70 ++++
.../pyflink/ml/recommendation/__init__.py | 17 +
.../pyflink/ml/recommendation/common.py | 48 +++
flink-ml-python/pyflink/ml/recommendation/swing.py | 222 +++++++++++
.../pyflink/ml/recommendation/tests/__init__.py | 30 ++
.../pyflink/ml/recommendation/tests/test_swing.py | 151 ++++++++
flink-ml-python/setup.py | 1 +
13 files changed, 1642 insertions(+)
diff --git a/docs/content/docs/operators/recommendation/_index.md b/docs/content/docs/operators/recommendation/_index.md
new file mode 100644
index 0000000..7b8f23e
--- /dev/null
+++ b/docs/content/docs/operators/recommendation/_index.md
@@ -0,0 +1,25 @@
+---
+title: Recommendation
+bookCollapseSection: true
+weight: 1
+aliases:
+ - /operators/recommendation/
+---
+<!--
+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.
+-->
diff --git a/docs/content/docs/operators/recommendation/swing.md b/docs/content/docs/operators/recommendation/swing.md
new file mode 100644
index 0000000..601ee82
--- /dev/null
+++ b/docs/content/docs/operators/recommendation/swing.md
@@ -0,0 +1,194 @@
+---
+title: "Swing"
+type: docs
+aliases:
+- /operators/recommendation/swing.html
+---
+
+<!--
+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.
+-->
+
+## Swing
+
+An AlgoOperator which implements the Swing algorithm.
+
+Swing is an item recall algorithm. 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.
+
+See "<a href="https://arxiv.org/pdf/2010.05525.pdf">Large Scale Product Graph Construction for Recommendation in
+E-commerce</a>" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+
+### Input Columns
+
+| Param name | Type | Default | Description |
+|:-----------|:-----|:---------|:------------|
+| itemCol | Long | `"item"` | Item id. |
+| userCol | Long | `"user"` | User id. |
+### Output Columns
+
+| Param name | Type | Default | Description |
+|:-----------|:-------|:-----------|:-----------------------------------------------------------------------------------------------|
+| itemCol | Long | `"item"` | Item id. |
+| outputCol | String | `"output"` | Top k similar items and their corresponding scores. (e.g. "item_1,0.9;item_2,0.7;item_3,0.35") |
+
+### Parameters
+
+Below are the parameters required by `Swing`.
+
+| Key | Default | Type | Required | Description |
+|:------------------|:-----------|:--------|:---------|:----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| userCol | `"user"` | String | no | User column name. |
+| itemCol | `"item"` | String | no | Item column name. |
+| maxUserNumPerItem | `1000` | Integer | no | The max number of user(purchasers) for each item. If the number of user is larger than this value, then only maxUserNumPerItem users will be sampled and considered in the computation of similarity between two items. |
+| k | `100` | Integer | no | The max number of similar items to output for each item. |
+| minUserBehavior | `10` | Integer | no | The min number of items for a user purchases. If the items purchased by a user is smaller than this value, then this user is filtered out while gathering data. This can affect the speed of the computation. Set minUserBehavior larger in case the swing recommendation progresses very slowly. |
+| maxUserBehavior | `1000` | Integer | no | The max number of items for a user purchases. If the items purchased by a user is larger than this value, then this user is filtered out while gathering data. This can affect the speed of the computation. Set maxUserBehavior smaller in case the swing recommendation progresses very slowly. The IllegalArgumentException is raised if the value of maxUserBehavior is smaller than minUserBehavior. |
+| alpha1 | `15` | Integer | no | Smooth factor for number of users that have purchased one item. The higher alpha1 is, the less purchasing behavior contributes to the similarity score. |
+| alpha2 | `0` | Integer | no | Smooth factor for number of users that have purchased the two target items. The higher alpha2 is, the less purchasing behavior contributes to the similarity score. |
+| beta | `0.3` | Double | no | Decay factor for number of users that have purchased one item. The higher beta is, the less purchasing behavior contributes to the similarity score. |
+| outputCol | `"output"` | String | no | Output column name. |
+
+### Examples
+
+{{< tabs examples >}}
+
+{{< tab "Java">}}
+
+```java
+package org.apache.flink.ml.examples.recommendation;
+
+import org.apache.flink.ml.recommendation.swing.Swing;
+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.flink.util.CloseableIterator;
+
+/**
+ * Simple program that creates a Swing instance and uses it to generate recommendations for items.
+ */
+public class SwingExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input data.
+ DataStream<Row> inputStream =
+ env.fromElements(
+ 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));
+
+ Table inputTable = tEnv.fromDataStream(inputStream).as("user", "item");
+
+ // Creates a Swing object and initializes its parameters.
+ Swing swing = new Swing().setUserCol("user").setItemCol("item").setMinUserBehavior(1);
+
+ // Transforms the data.
+ Table[] outputTable = swing.transform(inputTable);
+
+ // Extracts and displays the result of swing algorithm.
+ for (CloseableIterator<Row> it = outputTable[0].execute().collect(); it.hasNext(); ) {
+ Row row = it.next();
+
+ long mainItem = row.getFieldAs(0);
+ String itemRankScore = row.getFieldAs(1);
+
+ System.out.printf("item: %d, top-k similar items: %s\n", mainItem, itemRankScore);
+ }
+ }
+}
+
+```
+
+{{< /tab>}}
+
+{{< tab "Python">}}
+
+```python
+
+# 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.
+output_table = swing.transform(input_table)
+
+# Extracts and display the results.
+field_names = output_table[0].get_schema().get_field_names()
+
+results = t_env.to_data_stream(
+ output_table[0]).execute_and_collect()
+
+for result in results:
+ main_item = result[field_names.index(swing.get_item_col())]
+ item_rank_score = result[1]
+ print(f'item: {main_item}, top-k similar items: {item_rank_score}')
+
+```
+
+{{< /tab>}}
+
+{{< /tabs>}}
diff --git a/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/recommendation/SwingExample.java b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/recommendation/SwingExample.java
new file mode 100644
index 0000000..b75fb71
--- /dev/null
+++ b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/recommendation/SwingExample.java
@@ -0,0 +1,69 @@
+/*
+ * 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.examples.recommendation;
+
+import org.apache.flink.ml.recommendation.swing.Swing;
+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.flink.util.CloseableIterator;
+
+/**
+ * Simple program that creates a Swing instance and uses it to generate recommendations for items.
+ */
+public class SwingExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input data.
+ DataStream<Row> inputStream =
+ env.fromElements(
+ 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));
+
+ Table inputTable = tEnv.fromDataStream(inputStream).as("user", "item");
+
+ // Creates a Swing object and initializes its parameters.
+ Swing swing = new Swing().setUserCol("user").setItemCol("item").setMinUserBehavior(1);
+
+ // Transforms the data.
+ Table[] outputTable = swing.transform(inputTable);
+
+ // Extracts and displays the result of swing algorithm.
+ for (CloseableIterator<Row> it = outputTable[0].execute().collect(); it.hasNext(); ) {
+ Row row = it.next();
+
+ long mainItem = row.getFieldAs(0);
+ String itemRankScore = row.getFieldAs(1);
+
+ System.out.printf("item: %d, top-k similar items: %s\n", mainItem, itemRankScore);
+ }
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java
new file mode 100644
index 0000000..5a1ebc0
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/Swing.java
@@ -0,0 +1,417 @@
+/*
+ * 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.PrimitiveArrayTypeInfo;
+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.stream.Collectors;
+
+/**
+ * An AlgoOperator which implements the Swing algorithm.
+ *
+ * <p>Swing is an item recall algorithm. 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 similarity
+ * between items in Swing is defined as
+ *
+ * <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>Note that alpha1 and alpha2 could be zero here. If one of $$|I_u|, |I_v| and |I_u\cap I_v|$$
+ * is zero, then the similarity of <em>i</em> and <em>j</em> is zero.
+ *
+ * <p>See "<a href="https://arxiv.org/pdf/2010.05525.pdf">Large Scale Product Graph Construction for
+ * Recommendation in E-commerce</a>" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ */
+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) {
+ Preconditions.checkArgument(inputs.length == 1);
+ final String userCol = getUserCol();
+ final String itemCol = getItemCol();
+ 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 must be Long.");
+ }
+
+ if (getMaxUserBehavior() < getMinUserBehavior()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "The maxUserBehavior must be greater than or equal to minUserBehavior. "
+ + "The current setting: maxUserBehavior=%d, minUserBehavior=%d.",
+ getMaxUserBehavior(), getMinUserBehavior()));
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ SingleOutputStreamOperator<Tuple2<Long, Long>> purchasingBehavior =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ row -> {
+ Long userId = row.getFieldAs(userCol);
+ Long itemId = row.getFieldAs(itemCol);
+ if (userId == null || itemId == null) {
+ throw new RuntimeException(
+ "Data of user and item column must not be null.");
+ }
+ return Tuple2.of(userId, itemId);
+ })
+ .returns(Types.TUPLE(Types.LONG, Types.LONG));
+
+ SingleOutputStreamOperator<Tuple3<Long, Long, long[]>> userBehavior =
+ purchasingBehavior
+ .keyBy(tuple -> tuple.f0)
+ .transform(
+ "collectingUserBehavior",
+ Types.TUPLE(
+ Types.LONG,
+ Types.LONG,
+ PrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO),
+ new CollectingUserBehavior(
+ getMinUserBehavior(), getMaxUserBehavior()));
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ new TypeInformation[] {Types.LONG, Types.STRING},
+ new String[] {getItemCol(), getOutputCol()});
+
+ DataStream<Row> output =
+ userBehavior
+ .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);
+ }
+
+ /**
+ * Collects user behavior data and appends to the input table.
+ *
+ * <p>During the process, this operator collects users and all items he/she has purchased, and
+ * its input table must be bounded.
+ */
+ private static class CollectingUserBehavior
+ extends AbstractStreamOperator<Tuple3<Long, Long, long[]>>
+ implements OneInputStreamOperator<Tuple2<Long, Long>, Tuple3<Long, Long, long[]>>,
+ BoundedOneInput {
+ private final int minUserItemInteraction;
+ private final int maxUserItemInteraction;
+
+ // Maps a user id to a set of items. Because ListState cannot keep values of type `Set`,
+ // we use `Map<Long, String>` with null values instead. So does `userAndPurchasedItems` and
+ // `itemAndPurchasers` in `ComputingSimilarItems`.
+ private Map<Long, Map<Long, String>> userAndPurchasedItems = new HashMap<>();
+
+ private ListState<Map<Long, Map<Long, String>>> userAndPurchasedItemsState;
+
+ private CollectingUserBehavior(int minUserItemInteraction, int maxUserItemInteraction) {
+ this.minUserItemInteraction = minUserItemInteraction;
+ this.maxUserItemInteraction = maxUserItemInteraction;
+ }
+
+ @Override
+ public void endInput() {
+
+ userAndPurchasedItems.forEach(
+ (user, items) -> {
+ if (items.size() >= minUserItemInteraction
+ && items.size() <= maxUserItemInteraction) {
+ long[] itemsArray = new long[items.size()];
+ int i = 0;
+ for (Long value : items.keySet()) {
+ itemsArray[i++] = value;
+ }
+ items.forEach(
+ (item, nullValue) ->
+ output.collect(
+ new StreamRecord<>(
+ new Tuple3<>(user, item, itemsArray))));
+ }
+ });
+
+ userAndPurchasedItemsState.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 =
+ userAndPurchasedItems.getOrDefault(user, new LinkedHashMap<>());
+
+ if (items.size() <= maxUserItemInteraction) {
+ items.put(item, null);
+ }
+
+ userAndPurchasedItems.putIfAbsent(user, items);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ userAndPurchasedItemsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "userAndPurchasedItemsState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(
+ userAndPurchasedItemsState, "userAndPurchasedItemsState")
+ .ifPresent(stat -> userAndPurchasedItems = stat);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ userAndPurchasedItemsState.update(Collections.singletonList(userAndPurchasedItems));
+ }
+ }
+
+ /** Calculates similarity between items and keep top k similar items of each target item. */
+ private static class ComputingSimilarItems extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Tuple3<Long, Long, long[]>, Row>, BoundedOneInput {
+
+ private Map<Long, Map<Long, String>> userAndPurchasedItems = new HashMap<>();
+ private Map<Long, Map<Long, String>> itemAndPurchasers = new HashMap<>();
+ private ListState<Map<Long, Map<Long, String>>> userAndPurchasedItemsState;
+ private ListState<Map<Long, Map<Long, String>>> itemAndPurchasersState;
+
+ 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<>(userAndPurchasedItems.size());
+ userAndPurchasedItems.forEach(
+ (k, v) -> {
+ int count = v.size();
+ userWeights.put(k, calculateWeight(count));
+ });
+
+ for (long mainItem : itemAndPurchasers.keySet()) {
+ List<Long> userList =
+ sampleUserList(itemAndPurchasers.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 =
+ new HashSet<>(userAndPurchasedItems.get(u).keySet());
+ interaction.retainAll(userAndPurchasedItems.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());
+ String itemList =
+ itemAndScore.stream()
+ .sequential()
+ .limit(itemNums)
+ .map(tuple2 -> "" + tuple2.f0 + commaDelimiter + tuple2.f1)
+ .collect(Collectors.joining("" + semicolonDelimiter));
+ output.collect(new StreamRecord<>(Row.of(mainItem, itemList)));
+ }
+
+ userAndPurchasedItemsState.clear();
+ itemAndPurchasersState.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();
+ List<Long> userList = new ArrayList<>(allUsers.keySet());
+
+ if (totalSize < sampleSize) {
+ return userList;
+ }
+
+ Collections.shuffle(userList);
+ return userList.subList(0, sampleSize);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<Long, Long, long[]>> streamRecord)
+ throws Exception {
+ Tuple3<Long, Long, long[]> tuple3 = streamRecord.getValue();
+ long user = tuple3.f0;
+ long mainItem = tuple3.f1;
+ Map<Long, String> items = new HashMap<>();
+ for (long item : tuple3.f2) {
+ items.put(item, null);
+ }
+
+ userAndPurchasedItems.putIfAbsent(user, items);
+ itemAndPurchasers.putIfAbsent(mainItem, new HashMap<>());
+ itemAndPurchasers.get(mainItem).putIfAbsent(user, null);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ userAndPurchasedItemsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "userAndPurchasedItemsState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(
+ userAndPurchasedItemsState, "userAndPurchasedItemsState")
+ .ifPresent(stat -> userAndPurchasedItems = stat);
+
+ itemAndPurchasersState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "itemAndPurchasersState",
+ Types.MAP(
+ Types.LONG,
+ Types.MAP(Types.LONG, Types.STRING))));
+
+ OperatorStateUtils.getUniqueElement(itemAndPurchasersState, "itemAndPurchasersState")
+ .ifPresent(stat -> itemAndPurchasers = stat);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ userAndPurchasedItemsState.update(Collections.singletonList(userAndPurchasedItems));
+ itemAndPurchasersState.update(Collections.singletonList(itemAndPurchasers));
+ }
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/SwingParams.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/SwingParams.java
new file mode 100644
index 0000000..2f36c43
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/recommendation/swing/SwingParams.java
@@ -0,0 +1,168 @@
+/*
+ * 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.ml.common.param.HasOutputCol;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+import org.apache.flink.ml.param.WithParams;
+
+/**
+ * Params for {@link Swing}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface SwingParams<T> extends WithParams<T>, HasOutputCol<T> {
+ Param<String> USER_COL =
+ new StringParam("userCol", "User column name.", "user", ParamValidators.notNull());
+
+ Param<String> ITEM_COL =
+ new StringParam("itemCol", "Item column name.", "item", ParamValidators.notNull());
+
+ Param<Integer> MAX_USER_NUM_PER_ITEM =
+ new IntParam(
+ "maxUserNumPerItem",
+ "The max number of users(purchasers) for each item. If the number of users "
+ + "is greater than this value, then only maxUserNumPerItem users will "
+ + "be sampled and used in the computation of similarity between two items.",
+ 1000,
+ ParamValidators.gt(0));
+
+ Param<Integer> K =
+ new IntParam(
+ "k",
+ "The max number of similar items to output for each item.",
+ 100,
+ ParamValidators.gt(0));
+
+ Param<Integer> MIN_USER_BEHAVIOR =
+ new IntParam(
+ "minUserBehavior",
+ "The min number of items that a user purchases. If the items purchased by a user is smaller than "
+ + "this value, then this user is filtered out and will not be used in the computation.",
+ 10,
+ ParamValidators.gt(0));
+
+ Param<Integer> MAX_USER_BEHAVIOR =
+ new IntParam(
+ "maxUserBehavior",
+ "The max number of items for a user purchases. If the items purchased by a user is greater than "
+ + "this value, then this user is filtered out and will not be used in the computation.",
+ 1000,
+ ParamValidators.gt(0));
+
+ Param<Integer> ALPHA1 =
+ new IntParam(
+ "alpha1",
+ "Smooth factor for number of users that have purchased one item. The higher alpha1 is,"
+ + " the less purchasing behavior contributes to the similarity score.",
+ 15,
+ ParamValidators.gtEq(0));
+
+ Param<Integer> ALPHA2 =
+ new IntParam(
+ "alpha2",
+ "Smooth factor for number of users that have purchased the two target items. The higher alpha2 "
+ + "is, the less purchasing behavior contributes to the similarity score.",
+ 0,
+ ParamValidators.gtEq(0));
+
+ Param<Double> BETA =
+ new DoubleParam(
+ "beta",
+ "Decay factor for number of users that have purchased one item. The higher beta is, the less "
+ + "purchasing behavior contributes to the similarity score.",
+ 0.3,
+ ParamValidators.gtEq(0));
+
+ default String getUserCol() {
+ return get(USER_COL);
+ }
+
+ default T setUserCol(String value) {
+ return set(USER_COL, value);
+ }
+
+ default String getItemCol() {
+ return get(ITEM_COL);
+ }
+
+ default T setItemCol(String value) {
+ return set(ITEM_COL, value);
+ }
+
+ default int getK() {
+ return get(K);
+ }
+
+ default T setK(Integer value) {
+ return set(K, value);
+ }
+
+ default int getMaxUserNumPerItem() {
+ return get(MAX_USER_NUM_PER_ITEM);
+ }
+
+ default T setMaxUserNumPerItem(Integer value) {
+ return set(MAX_USER_NUM_PER_ITEM, value);
+ }
+
+ default int getMinUserBehavior() {
+ return get(MIN_USER_BEHAVIOR);
+ }
+
+ default T setMinUserBehavior(Integer value) {
+ return set(MIN_USER_BEHAVIOR, value);
+ }
+
+ default int getMaxUserBehavior() {
+ return get(MAX_USER_BEHAVIOR);
+ }
+
+ default T setMaxUserBehavior(Integer value) {
+ return set(MAX_USER_BEHAVIOR, value);
+ }
+
+ default int getAlpha1() {
+ return get(ALPHA1);
+ }
+
+ default T setAlpha1(Integer value) {
+ return set(ALPHA1, value);
+ }
+
+ default int getAlpha2() {
+ return get(ALPHA2);
+ }
+
+ default T setAlpha2(Integer value) {
+ return set(ALPHA2, value);
+ }
+
+ default double getBeta() {
+ return get(BETA);
+ }
+
+ default T setBeta(Double value) {
+ return set(BETA, value);
+ }
+}
diff --git a/flink-ml-lib/src/test/java/org/apache/flink/ml/recommendation/SwingTest.java b/flink-ml-lib/src/test/java/org/apache/flink/ml/recommendation/SwingTest.java
new file mode 100644
index 0000000..8c17fdd
--- /dev/null
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/recommendation/SwingTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.apache.commons.lang3.exception.ExceptionUtils;
+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.Comparator;
+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 inputTable;
+
+ @Before
+ public void before() {
+ env = TestUtils.getExecutionEnvironment();
+ tEnv = StreamTableEnvironment.create(env);
+ List<Row> inputRows =
+ 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),
+ Row.of(4L, 12L),
+ Row.of(4L, 10L),
+ Row.of(4L, 11L),
+ Row.of(4L, 12L),
+ Row.of(4L, 13L)));
+ inputTable =
+ tEnv.fromDataStream(
+ env.fromCollection(
+ inputRows,
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO,
+ BasicTypeInfo.LONG_TYPE_INFO
+ },
+ new String[] {"user", "item"})));
+ }
+
+ private void compareResultAndExpected(List<Row> results) {
+ 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")));
+
+ results.sort(Comparator.comparing(o -> o.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);
+ assertEquals(expect.getField(0), result.getField(0));
+ assertEquals(expect.getField(1), itemRankScore);
+ }
+ }
+
+ @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 testInputWithIllegalDataType() {
+ 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", "item"}));
+ Table data = tEnv.fromDataStream(dataStream);
+
+ try {
+ Table[] swingResultTables = new Swing().setMinUserBehavior(1).transform(data);
+ swingResultTables[0].execute().collect();
+ fail();
+ } catch (RuntimeException e) {
+ assertEquals(IllegalArgumentException.class, e.getClass());
+ assertEquals("The types of user and item must be Long.", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testInputWithNull() {
+ 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,
+ new RowTypeInfo(
+ new TypeInformation[] {
+ BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
+ },
+ new String[] {"user", "item"}));
+ Table data = tEnv.fromDataStream(dataStream);
+ Swing swing = new Swing().setMinUserBehavior(1);
+ Table[] swingResultTables = swing.transform(data);
+
+ try {
+ swingResultTables[0].execute().collect().next();
+ fail();
+ } catch (RuntimeException e) {
+ Throwable exception = ExceptionUtils.getRootCause(e);
+ assertEquals(RuntimeException.class, exception.getClass());
+ assertEquals("Data of user and item column must not be null.", exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testOutputSchema() {
+ Swing swing = new Swing().setOutputCol("item_score").setMinUserBehavior(1);
+ Table[] swingResultTables = swing.transform(inputTable);
+ Table output = swingResultTables[0];
+
+ assertEquals(
+ Arrays.asList("item", "item_score"), output.getResolvedSchema().getColumnNames());
+ }
+
+ @Test
+ public void testTransform() {
+ Swing swing = new Swing().setMinUserBehavior(2).setMaxUserBehavior(3);
+ Table[] swingResultTables = swing.transform(inputTable);
+ Table outputTable = swingResultTables[0];
+ List<Row> results = IteratorUtils.toList(outputTable.execute().collect());
+ compareResultAndExpected(results);
+ }
+
+ @Test
+ public void testSaveLoadAndTransform() throws Exception {
+ Swing swing = new Swing().setMinUserBehavior(1);
+ Swing loadedSwing =
+ TestUtils.saveAndReload(tEnv, swing, tempFolder.newFolder().getAbsolutePath());
+ Table outputTable = loadedSwing.transform(inputTable)[0];
+ List<Row> results = IteratorUtils.toList(outputTable.execute().collect());
+ compareResultAndExpected(results);
+ }
+}
diff --git a/flink-ml-python/pyflink/examples/ml/recommendation/swing_example.py b/flink-ml-python/pyflink/examples/ml/recommendation/swing_example.py
new file mode 100644
index 0000000..4ba6684
--- /dev/null
+++ b/flink-ml-python/pyflink/examples/ml/recommendation/swing_example.py
@@ -0,0 +1,70 @@
+################################################################################
+# 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.
+output_table = swing.transform(input_table)
+
+# Extracts and display the results.
+field_names = output_table[0].get_schema().get_field_names()
+
+results = t_env.to_data_stream(
+ output_table[0]).execute_and_collect()
+
+for result in results:
+ main_item = result[field_names.index(swing.get_item_col())]
+ item_rank_score = result[1]
+ print(f'item: {main_item}, top-k similar items: {item_rank_score}')
diff --git a/flink-ml-python/pyflink/ml/recommendation/__init__.py b/flink-ml-python/pyflink/ml/recommendation/__init__.py
new file mode 100644
index 0000000..65b48d4
--- /dev/null
+++ b/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.
+################################################################################
diff --git a/flink-ml-python/pyflink/ml/recommendation/common.py b/flink-ml-python/pyflink/ml/recommendation/common.py
new file mode 100644
index 0000000..7a6142d
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/recommendation/common.py
@@ -0,0 +1,48 @@
+################################################################################
+# 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 abc import ABC, abstractmethod
+
+from pyflink.ml.wrapper import JavaAlgoOperator
+
+JAVA_RECOMMENDATION_PACKAGE_NAME = "org.apache.flink.ml.recommendation"
+
+
+class JavaRecommendationAlgoOperator(JavaAlgoOperator, ABC):
+ """
+ Wrapper class for a Java Recommendation AlgoOperator.
+ """
+
+ def __init__(self, java_algo_operator):
+ super(JavaRecommendationAlgoOperator, self).__init__(java_algo_operator)
+
+ @classmethod
+ def _java_stage_path(cls) -> str:
+ return ".".join(
+ [JAVA_RECOMMENDATION_PACKAGE_NAME,
+ cls._java_algo_operator_package_name(),
+ cls._java_algo_operator_class_name()])
+
+ @classmethod
+ @abstractmethod
+ def _java_algo_operator_package_name(cls) -> str:
+ pass
+
+ @classmethod
+ @abstractmethod
+ def _java_algo_operator_class_name(cls) -> str:
+ pass
diff --git a/flink-ml-python/pyflink/ml/recommendation/swing.py b/flink-ml-python/pyflink/ml/recommendation/swing.py
new file mode 100644
index 0000000..41fa1b8
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/recommendation/swing.py
@@ -0,0 +1,222 @@
+################################################################################
+# 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 JavaRecommendationAlgoOperator
+from pyflink.ml.wrapper import JavaWithParams
+
+
+class _SwingParams(
+ JavaWithParams,
+ HasOutputCol
+):
+ """
+ Params for :class:`Swing`.
+ """
+
+ USER_COL: Param[str] = StringParam(
+ "user_col",
+ "User column name.",
+ "user",
+ ParamValidators.not_null())
+
+ ITEM_COL: Param[str] = StringParam(
+ "item_col",
+ "Item column name.",
+ "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(purchasers) for each item. If the number of users "
+ + "is greater than this value, then only maxUserNumPerItem users will "
+ + "be sampled and used in the computation of similarity between two items.",
+ 1000,
+ ParamValidators.gt(0))
+
+ MIN_USER_BEHAVIOR: Param[int] = IntParam(
+ "min_user_behavior",
+ "The min number of items that a user purchases. If the items purchased by a user is "
+ + "smaller than this value, then this user is filtered out and will not be used in the "
+ + "computation.",
+ 10,
+ ParamValidators.gt(0))
+
+ MAX_USER_BEHAVIOR: Param[int] = IntParam(
+ "max_user_behavior",
+ "The max number of items for a user purchases. If the items purchased by a user is "
+ + "greater than this value, then this user is filtered out and will not be used in the "
+ + "computation.",
+ 1000,
+ ParamValidators.gt(0))
+
+ ALPHA1: Param[int] = IntParam(
+ "alpha1",
+ "Smooth factor for number of users that have purchased one item. The higher alpha1 is,"
+ + " the less purchasing behavior contributes to the similarity score.",
+ 15,
+ ParamValidators.gt_eq(0))
+
+ ALPHA2: Param[int] = IntParam(
+ "alpha2",
+ "Smooth factor for number of users that have purchased the two target items. The higher "
+ + "alpha2 is, the less purchasing behavior contributes to the similarity score.",
+ 0,
+ ParamValidators.gt_eq(0))
+
+ BETA: Param[float] = FloatParam(
+ "beta",
+ "Decay factor for number of users that have purchased one item. The higher beta is, the "
+ + "less purchasing behavior contributes to the similarity score.",
+ 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(JavaRecommendationAlgoOperator, _SwingParams):
+ """
+ An AlgoOperator which implements the Swing algorithm.
+
+ Swing is an item recall algorithm. 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 similarity
+ between items in Swing is defined as
+
+ $$ 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|}} $$
+
+ Note that alpha1 and alpha2 could be zero here. If one of $$|I_u|, |I_v| and |I_u\\cap I_v|$$
+ is zero, then the similarity of <em>i</em> and <em>j</em> is zero.
+
+ See "<a href="https://arxiv.org/pdf/2010.05525.pdf">Large Scale Product Graph Construction for
+ Recommendation in E-commerce</a>" by Xiaoyong Yang, Yadong Zhu and Yi Zhang.
+ """
+
+ def __init__(self, java_algo_operator=None):
+ super(Swing, self).__init__(java_algo_operator)
+
+ @classmethod
+ def _java_algo_operator_package_name(cls) -> str:
+ return "swing"
+
+ @classmethod
+ def _java_algo_operator_class_name(cls) -> str:
+ return "Swing"
diff --git a/flink-ml-python/pyflink/ml/recommendation/tests/__init__.py b/flink-ml-python/pyflink/ml/recommendation/tests/__init__.py
new file mode 100644
index 0000000..6698191
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/recommendation/tests/__init__.py
@@ -0,0 +1,30 @@
+################################################################################
+# 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 os
+import sys
+from pathlib import Path
+
+# Because the project and the dependent `pyflink` project have the same directory structure,
+# we need to manually add `flink-ml-python` path to `sys.path` in the test of this project to change
+# the order of package search.
+flink_ml_python_dir = Path(__file__).parents[5]
+sys.path.append(str(flink_ml_python_dir))
+
+import pyflink
+
+pyflink.__path__.insert(0, os.path.join(flink_ml_python_dir, 'pyflink'))
diff --git a/flink-ml-python/pyflink/ml/recommendation/tests/test_swing.py b/flink-ml-python/pyflink/ml/recommendation/tests/test_swing.py
new file mode 100644
index 0000000..daa9357
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/recommendation/tests/test_swing.py
@@ -0,0 +1,151 @@
+################################################################################
+# 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 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.input_table = 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(1000, swing.max_user_num_per_item)
+ self.assertEqual(10, swing.min_user_behavior)
+ self.assertEqual(1000, swing.max_user_behavior)
+ self.assertEqual(15, swing.alpha1)
+ self.assertEqual(0, swing.alpha2)
+ self.assertAlmostEqual(0.3, swing.beta, delta=1e-9)
+
+ swing.set_item_col("item_1") \
+ .set_user_col("user_1") \
+ .set_k(20) \
+ .set_max_user_num_per_item(500)\
+ .set_min_user_behavior(20) \
+ .set_max_user_behavior(50) \
+ .set_alpha1(5) \
+ .set_alpha2(1) \
+ .set_beta(0.35)
+
+ self.assertEqual("item_1", swing.item_col)
+ self.assertEqual("user_1", swing.user_col)
+ self.assertEqual(20, swing.k)
+ self.assertEqual(500, swing.max_user_num_per_item)
+ self.assertEqual(20, swing.min_user_behavior)
+ self.assertEqual(50, swing.max_user_behavior)
+ self.assertEqual(5, swing.alpha1)
+ self.assertEqual(1, swing.alpha2)
+ self.assertAlmostEqual(0.35, swing.beta, delta=1e-9)
+
+ def test_output_schema(self):
+ swing = Swing() \
+ .set_item_col('test_item') \
+ .set_user_col('test_user') \
+ .set_output_col("item_score")
+
+ output = swing.transform(self.input_table.alias('test_user', 'test_item'))[0]
+ self.assertEqual(
+ ['test_item', 'item_score'],
+ output.get_schema().get_field_names())
+
+ def test_transform(self):
+ swing = Swing().set_min_user_behavior(1)
+ output = swing.transform(self.input_table)[0]
+ self.verify_output_result(
+ output,
+ swing.get_item_col(),
+ output.get_schema().get_field_names(),
+ self.expected_data)
+
+ def test_save_load_and_transform(self):
+ swing = Swing().set_min_user_behavior(1)
+ reloaded_swing = self.save_and_reload(swing)
+ output = reloaded_swing.transform(self.input_table)[0]
+ self.verify_output_result(
+ output,
+ swing.get_item_col(),
+ output.get_schema().get_field_names(),
+ self.expected_data)
+
+ def verify_output_result(
+ self, output: Table,
+ item_col: str,
+ field_names: List[str],
+ expected_result: List):
+ collected_results = [result for result in
+ self.t_env.to_data_stream(output).execute_and_collect()]
+ results = []
+ for result in collected_results:
+ main_item = result[field_names.index(item_col)]
+ item_rank_score = result[1]
+ results.append([main_item, item_rank_score])
+ results.sort(key=lambda x: x[0])
+ self.assertEqual(expected_result, results)
diff --git a/flink-ml-python/setup.py b/flink-ml-python/setup.py
index fc234c4..6c67c9f 100644
--- a/flink-ml-python/setup.py
+++ b/flink-ml-python/setup.py
@@ -91,6 +91,7 @@ try:
'pyflink.ml.clustering',
'pyflink.ml.evaluation',
'pyflink.ml.feature',
+ 'pyflink.ml.recommendation',
'pyflink.ml.regression',
'pyflink.ml.stats',
'pyflink.ml.util',