阿里云推荐引擎使用调查记录

深碍√TFBOYSˉ_ 2022-06-13 05:49 162阅读 0赞

最近项目中需要使用到推荐引擎，直接选用的是阿里云推荐引擎。由于一开始都没涉及过这方面的知识，而且调查时正好是V3发布前，调查到一半的时候，推荐引擎发布了最新V3版本，DDL和V2的略有不同。在调查与开发过程中踩了不少的坑，这里总结下。

首先最有感触的就是阿里的工单真好用，回复的效率还是很高的，基本上当天就能得到有效的回复。所以开发中很多问题和解决方法都是从咨询客服中获取的。

1、已得知的基础情报

1、推荐模块下的api调用目前官方没有提供SDK，需要自己使用API进行调用。
    2、大数据模块下的官方提供了SDK，地址：https://help.aliyun.com/document_detail/34614.html
    3、一个project其实是一个数据库资源，为不同的业务提供基础数据
    4、业务依赖依赖数据库资源，不同场景可以选择不同算法
    5、SDK使用说明：https://help.aliyun.com/document_detail/34614.html
    6、API调用demo:https://help.aliyun.com/document_detail/30245.html
    7、在推荐引擎管理后台上，可以编写脚本和启动定时任务，目前还么使用到

2、踩坑点

1、demo上用的是V2版本的数据和DDL，我们使用V3版本的DDL进行初始化，算法任务死活是失败的。
    2、V3版本刚发布时，算法任务失败log是看不到的，总是提示队列处理中。没办法，项目比较紧急，只能选用V2版本进行开发。
    3、使用SDK上传数据的时候，以为只需要把需要用到的字段传入数据即可，测试后发现数据导入的字段不对。需要把表结构上的所有字段按顺序定义出来，然后再根据字段名称进行赋值

--------------------

以下为使用到的基本代码，部分从官方文档中获取，部分是咨询客服获取到的。  
1、初始化API账号数据

Account account = new AliyunAccount(accessId, accessKey);
                Odps odps = new Odps(account);
                odps.setEndpoint(endPoint);
                odps.setDefaultProject(project);
    
                TableTunnel tunnel = new TableTunnel(odps);
                tunnel.setEndpoint("http://dt.odps.aliyun.com");
    从官网：https://help.aliyun.com/document_detail/30245.html 中获取的代码片段

2、判断分区是否存在，不存在则建立分区

public boolean createPartitionIfNeed(String dsDate, Odps odps, String table) throws OdpsException {
                Tables tbs = odps.tables();
                Table t = tbs.get(table);
                String partition = "ds=" + dsDate;
                PartitionSpec partitionSpec = new PartitionSpec(partition);
    
                boolean partitionExists = false;
                for (Partition p : t.getPartitions()) {
                      if (StringUtils.equals(p.getPartitionSpec().get("ds"), dsDate)) {
                            partitionExists = true;
                            break;
                      }
                }
    
                if (!partitionExists) {
                      logger.info("分区{}不存在，新建分区！", dsDate);
                      t.createPartition(partitionSpec);
                      logger.info("分区{}不存在，新建分区成功！", dsDate);
                }
    
                return !partitionExists;
          }

3、启动数据预处理任务  
4、启动离线算法任务  
5、查询任务状态

目前官方没有提供SDK，需要通过API调用

6、执行SQL,从dual表中获取数据

public class RecTest { 
        private static final String accessId = "";
        private static final String accessKey = "";
        private static final String endPoint = "http://service.odps.aliyun.com/api";
        private static final String project = "myproject";
        private static final String ct = "20170624";
        private static final String sql = "INSERT OVERWRITE TABLE demo.demo_user_meta PARTITION (ds="+ ct +") SELECT x ,null FROM demo.dual_demo WHERE x = '20170620';";
    
        public static void main(String[] args) {
            Account account = new AliyunAccount(accessId, accessKey);
            Odps odps = new Odps(account);
            odps.setEndpoint(endPoint);
            odps.setDefaultProject(project);
            Instance i;
            try {
                i = SQLTask.run(odps, sql);
                i.waitForSuccess();
                List<Record> records = SQLTask.getResult(i);
                for (Record r : records) {
                    System.out.println(r.get(0).toString());
                }
            } catch (OdpsException e) {
                e.printStackTrace();
            }
        }
    }

7、上传数据

UploadSession uploadSession = tunnel.createUploadSession(project, table, new PartitionSpec("ds=20170623"));
                RecordWriter recordWriter = uploadSession.openRecordWriter(0);
    
                String date = "20170623";
    
                for (int i = 0; i < 60; i++) {
                      //这里需要注意，需要把表结构下的所有字段按表结构顺序定义出来，否则会出现数据导到其他字段中去的情况
                      Record record = new ArrayRecord(new Column[] { new Column("item_id", OdpsType.STRING), new Column("ds", OdpsType.STRING),
                                  new Column("category", OdpsType.STRING) });
                      record.set("item_id", "itemid_1" + i);
                      record.set("ds", date);
                      record.set("category", "category" + i);
                      recordWriter.write(record);
                }
    
    
                recordWriter.close();
    
                long id = uploadSession.getAvailBlockId();
    
                uploadSession.commit(new Long[] { id });

8、获取表数据

DownloadSession downloadSession = tunnel.createDownloadSession(project, table, new PartitionSpec("ds=20170623"));
                long count = downloadSession.getRecordCount();
                RecordReader recordReader = downloadSession.openRecordReader(0, count);
                Record record;
    
                while ((record = recordReader.read()) != null) {
                      int colCount = record.getColumnCount();
    
                      for (int i = 0; i < colCount; i++) { 
                            System.out.print(record.get(i) + ",");
                      }
                      System.out.println();
                }
    
                recordReader.close();
    
                System.out.println("总数量为：" + count);

通过以上的接口功能，推荐功能的基本步骤如下：  
1、同过SDK上传完用户基本数据和行为数据  
2、 调用API启动数据预处理任务，获取到任务id  
3、根据任务id判断数据预处理是否成功  
4、数据预处理成功后调用API启动离线算法，获取到任务id  
5、根据步骤4获取到的任务id，判断离线算法是否处理成功