MaxCompute实践之路（三） -- Java对接MaxCompute

一. jdbc方式对接maxcompute

创建完成项目后，创建以下类，根据注释，调好对应参数即可

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;public class Main {private static final String DRIVER_NAME = "com.aliyun.odps.jdbc.OdpsDriver";public static void main(String[] args) throws SQLException {try {Class.forName(DRIVER_NAME);} catch (ClassNotFoundException e) {e.printStackTrace();System.exit(1);}Connection conn = DriverManager.getConnection("jdbc:odps:<maxcompute endpoint>?project=<maxcompute project>","aliyun accessId", "aliyun accessKey");ResultSet rs;Statement stmt = conn.createStatement();String sql = "SELECT * FROM JDBC_TEST";stmt.executeQuery(sql);ResultSet rset = stmt.getResultSet();while (rset.next()) {System.out.println(String.valueOf(rset.getInt(1)) + "\t" + rset.getString(2));}}
}

项目结构如下

查看官方文档中的生态对接-jdbc参考-使用说明。
在上面创建的module项目的java包中创建测试类，参考官网例子即可。
官网链接：https://help.aliyun.com/document_detail/177015.html

二. java jdk中的SQLTask方式对接maxcompute

过程与jdbc的相同，只需重新创建以下实体类即可

import java.util.List;
import com.aliyun.odps.Instance;
import com.aliyun.odps.Odps;
import com.aliyun.odps.OdpsException;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.task.SQLTask;
public class testSql {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 = "";private static final String sql = "select category from iris;";public static voidmain(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();}}
}

官网链接：
https://help.aliyun.com/document_detail/34614.html?spm=a2c4g.11186623.6.670.3b0c4849C06sIW

三. 流式数据通道SDK（Streaming Tunnel）方式对接maxcompute

MaxCompute流式数据通道服务提供了以流式的方式把数据写入MaxCompute的能力，使用与原批量数据通道服务不同的一套全新的API及后端服务。流式服务在API上极大简化了分布式服务的开发成本，同时解决了批量数据通道在高并发、高QPS（Queries-per-second）场景下的性能瓶颈。

对应该方式实体类。

import java.io.IOException;
import java.util.Date;
import com.aliyun.odps.Column;
import com.aliyun.odps.Odps;
import com.aliyun.odps.PartitionSpec;
import com.aliyun.odps.TableSchema;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.tunnel.TunnelException;
public class StreamUploadSample {// 阿里云账号AccessKey ID。private static String accessId = "<your_access_id>";// 阿里云账号AccessKey Secret。private static String accessKey = "<your_access_key>";// MaxCompute项目的Endpoint信息，详情请参见Endpoint。private static String odpsEndpoint = "<endpoint>";// MaxCompute项目的Tunnel Endpoint信息，详情请参见Endpoint。private static String tunnelEndpoint = "<tunnel_endpoint>";// MaxCompute项目的名称。private static String project = "<your_project>";// MaxCompute项目中的表名称。private static String table = "<your_table_name>";// MaxCompute项目中的表的分区信息。private static String partition = "<your_partition_spec>";public static void main(String args[]) {Account account = new AliyunAccount(accessId, accessKey);Odps odps = new Odps(account);odps.setEndpoint(odpsEndpoint);odps.setDefaultProject(project);try {TableTunnel tunnel = new TableTunnel(odps);// 默认情况下，不需要设置Tunnel Endpoint，可以通过Endpoint自动路由。// 只有少数场景（比如路由的Endpoint网络不通）需要设置Tunnel Endpoint。// 可以通过以下接口设置Tunnel Endpoint。// tunnel.setEndpoint(tunnelEndpoint);PartitionSpec partitionSpec = new PartitionSpec(partition);TableTunnel.StreamUploadSession uploadSession = tunnel.createStreamUploadSession(project,table, partitionSpec);TableSchema schema = uploadSession.getSchema();TableTunnel.StreamRecordPack pack = uploadSession.newRecordPack();Record record = uploadSession.newRecord();for (int i = 0; i < schema.getColumns().size(); i++) {Column column = schema.getColumn(i);switch (column.getType()) {case BIGINT:record.setBigint(i, 1L);break;case BOOLEAN:record.setBoolean(i, true);break;case DATETIME:record.setDatetime(i, new Date());break;case DOUBLE:record.setDouble(i, 0.0);break;case STRING:record.setString(i, "sample");break;default:throw new RuntimeException("Unknown column type: "+ column.getType());}}for (int i = 0; i < 10; i++) {pack.append(record);}int retry = 0;while (retry < 3) {try {// flush成功表示数据写入成功，写入成功后数据立即可见。// flush成功后pack对象可以复用，避免频繁申请内存导致内存回收。// flush失败可以直接重试。// flush失败后pack对象不可重用，需要重新创建新的StreamRecordPack对象。String traceId = pack.flush();System.out.println("flush success:" + traceId);break;} catch (IOException e) {retry++;e.printStackTrace();Thread.sleep(500);}}System.out.println("upload success!");} catch (TunnelException e) {e.printStackTrace();} catch (IOException | InterruptedException e) {e.printStackTrace();}}
}

使用经验：
该方式可以在创建的pick中插入多个Record进行提交，例如插入10000个。经过测试pick中1条数据与10000条数据上传到库里的时间基本没差距。
应对高并发情况，可将创建链接的代码抽离成工具类，使其不用每次提交都创建链接，可提高上传速度。
创建表时，加入分区字段，可提高查询速度。
官网链接：
https://help.aliyun.com/document_detail/198167.html

MaxCompute实践之路（三） -- Java对接MaxCompute相关推荐

领域驱动设计(DDD)实践之路(三)：如何设计聚合
本文首发于 vivo互联网技术微信公众号链接:https://mp.weixin.qq.com/s/oAD25H0UKH4zujxFDRXu9Q 作者:wenbo zhang [领域驱动设计实践 ...
Java程序员必经的实践之路：Java中高级核心知识全面解析(11)
ThreadPoolTaskExecutor定义一些策略: ThreadPoolExecutor.AbortPolicy:抛出RejectedExecutionException来拒绝新任务的处理. ...
Java程序员必经的实践之路：java控制台在哪里打开
前言作为一名编程人员,对MySQL一定不会陌生,尤其是互联网行业,对MySQL的使用是比较多的.对于求职者来说,MySQL又是面试中一定会问到的重点,很多人拥有大厂梦,却因为MySQL败下阵来.实际 ...
软件性能测试分析与调优实践之路-Java应用程序的性能分析与调优-手稿节选
Java编程语言自从诞生起,就成为了一门非常流行的编程语言,覆盖了互联网.安卓应用.后端应用.大数据等很多技术领域,因此Java应用程序的性能分析和调优也是一门非常重要的课题.Java应用程序的性能直 ...
《深入理解Java虚拟机 - Jvm高级特性与最佳实践（第三版）》阅读笔记
<深入理解Java虚拟机>阅读笔记本repository为<深入理解Java虚拟机 - Jvm高级特性与最佳实践(第三版)>阅读笔记,因为第一章主要讲的是Java的发展历史, ...
java对接微信支付收不到支付通知问题(亲身实践)
问题描述: 用java对接微信支付时,统一下单接口正常.但是用户扫码付款成功后,设置用于回调的notify_url对应的接口并没有收到请求(这个url测试过,是正常的且外网能访问的). 由于官方文档没 ...
maxcompute mysql_MySQL/RDS数据如何同步到MaxCompute之实践讲解
实验方案概述本实验是对RDS同步数据到MaxCompute的一个初步讲解.当企业需要利用MaxCompute进行数据开发时,如果数据不在MaxCompute而在RDS中,首先需要将RDS中的数据同步 ...
猪八戒网CI/CD最佳实践之路
序言本文旨在介绍ZBJ DevOps团队倾力打造的DevOps平台中关于CI/CD流水线部分的实践.历经三次大版本迭代更新的流水线,完美切合ZBJ各种业务发展需求,在满足高频率交付的同时,提高了研发 ...
云效助力新金融DevOps转型——南京银行实践之路
在2018云栖大会南京峰会企业研发云专场,由南京银行研发管理负责人吴攀带来了"云效助力新金融DevOps转型--南京银行实践之路"的主题分享.首先对南京银行的研发规模与成长做了介绍 ...

MaxCompute实践之路（三） -- Java对接MaxCompute

一. jdbc方式对接maxcompute

二. java jdk中的SQLTask方式对接maxcompute

三. 流式数据通道SDK（Streaming Tunnel）方式对接maxcompute

MaxCompute实践之路（三） -- Java对接MaxCompute相关推荐

最新文章

热门文章