FlattenKObject (FlattenKObjectStreamOp)

Java 类名：com.alibaba.alink.operator.stream.recommendation.FlattenKObjectStreamOp
Python 类名：FlattenKObjectStreamOp

功能介绍

将流式推荐结果从json序列化格式转为table格式。

参数说明

代码示例

Python 代码

from pyalink.alink import *
import pandas as pd
useLocalEnv(1)
df_data = pd.DataFrame([
    [1,'{"rating":"[0.6]","object":"[1]"}'],
    [2,'{"rating":"[0.8,0.6]","object":"[2,3]"}'],
    [3,'{"rating":"[0.6,0.3,0.4]","object":"[1,2,3]"}']
])
data = BatchOperator.fromDataframe(df_data, schemaStr='user bigint, rec string')
sdata = StreamOperator.fromDataframe(df_data, schemaStr='user bigint, rec string')
recList = FlattenKObjectStreamOp()\
            .setSelectedCol("rec")\
            .setOutputCols(["object", "rating"])\
            .setOutputColTypes(["long", "double"])\
            .setReservedCols(["user"])\
            .linkFrom(sdata).print();
StreamOperator.execute()

Java 代码

import org.apache.flink.types.Row;
import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.recommendation.FlattenKObjectStreamOp;
import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
import org.junit.Test;
import java.util.Arrays;
import java.util.List;
public class FlattenKObjectStreamOpTest {
    @Test
    public void testFlattenKObjectStreamOp() throws Exception {
        List <Row> df_data = Arrays.asList(
            Row.of(1, "{\"rating\":\"[0.6]\",\"object\":\"[1]\"}"),
            Row.of(2, "{\"rating\":\"[0.8,0.6]\",\"object\":\"[2,3]\"}"),
            Row.of(3, "{\"rating\":\"[0.6,0.3,0.4]\",\"object\":\"[1,2,3]\"}")
        );
        BatchOperator <?> data = new MemSourceBatchOp(df_data, "user int, rec string");
        StreamOperator <?> sdata = new MemSourceStreamOp(df_data, "user int, rec string");
        StreamOperator <?> recList = new FlattenKObjectStreamOp()
            .setSelectedCol("rec")
            .setOutputCols("object", "rating")
            .setOutputColTypes("long", "double")
            .setReservedCols("user")
            .linkFrom(sdata).print();
        StreamOperator.execute();
    }
}

运行结果