MaxCompute Tunnel SDK数据上传利器——BufferedWriter使用指南

简介: MaxCompute 的数据上传接口(Tunnel)定义了数据 block 的概念:一个 block 对应一个 http request,多个 block 的上传可以并发而且是原子的,一次同步请求要么成功要么失败,不会污染其他的 block。这种设计对于服务端来讲十分简洁,但是也把记录状态做 fa.

本文用到的

阿里云数加-大数据计算服务MaxCompute产品地址:https://www.aliyun.com/product/odps


MaxCompute 的数据上传接口(Tunnel)定义了数据 block 的概念:一个 block 对应一个 http request,多个 block 的上传可以并发而且是原子的,一次同步请求要么成功要么失败,不会污染其他的 block。这种设计对于服务端来讲十分简洁,但是也把记录状态做 failover 的工作交给了客户端。

用户在使用 Tunnel SDK 编程时,需要对 block 这一层的语义进行认知,并且驱动数据上传的整个过程[1],并且自己进行容错,毕竟『网络错误是正常而不是异常』。由于用户文档中并没有强调这一点的重要性,导致很多用户踩了坑,一种常见的出错场景是,当客户端写数据的速度过慢,两次 write 的间隔超时[2],导致整个 block 上传失败。

High Level API

MaxCompute Java SDK 在 0.21.3-public  之后新增了 BufferredWriter 这个更高层的 API,简化了数据上传的过程,并且提供了容错的功能。 BufferedWriter 对用户隐藏了 block 这个概念,从用户角度看,就是在 session 上打开一个 writer 然后往里面写记录即可:

RecordWriter writer = null;

try {
  int i = 0;  
  writer = uploadSession.openBufferedWriter();
  Record product = uploadSession.newRecord();

  for (String item : items) {
    product.setString("name", item);
    product.setBigint("id", i);
    writer.write(product);
    i += 1;
  }
} finally {
  if (writer != null) {
    writer.close();
  }
}
uploadSession.commit();

具体实现时 BufferedWriter 先将记录缓存在客户端的缓冲区中,并在缓冲区填满之后打开一个 http 连接进行上传。BufferedWriter 会尽最大可能容错,保证数据上传上去。

  • 由于屏蔽了底层细节,这个接口可能并不适合数据预划分、断点续传、分批次上传等需要细粒度控制的场景。

多线程上传示例

多线程上传时,每个线程只需要打开一个 writer 往里面写数据就行了。

class UploadThread extends Thread {
  private UploadSession session;
  private static int RECORD_COUNT = 1200;

  public UploadThread(UploadSession session) {
    this.session = session;
  }

  @Override
  public void run() {
    RecordWriter writer = up.openBufferedWriter();
    Record r = up.newRecord();
    for (int i = 0; i < RECORD_COUNT; i++) {
      r.setBigint(0, i);
      writer.write(r);
    }
    writer.close();
  }
};

public class Example {
  public static void main(String args[]) {

   // 初始化 MaxCompute 和 tunnel 的代码

   TableTunnel.UploadSession uploadSession = tunnel.createUploadSession(projectName, tableName);
   UploadThread t1 = new UploadThread(up);
   UploadThread t2 = new UploadThread(up);

   t1.start();
   t2.start();
   t1.join();
   t2.join();

   uploadSession.commit();
 }

更多控制

重试策略

由于底层在上传出错时会回避一段固定的时间并进行重试,但如果你的程序不想花太多时间在重试上,或者你的程序位于一个极其恶劣的网络环境中,为此 TunnelBufferedWriter 允许用户配置重试策略。

用户可以选择三种重试回避策略:指数回避(EXPONENTIAL_BACKOFF)、线性时间回避(LINEAR_BACKOFF)、常数时间回避(CONSTANT_BACKOFF)。

例如下面这段代码可以将,write 的重试次数调整为 6,每一次重试之前先分别回避 4s、8s、16s、32s、64s 和 128s(从 4 开始的指数递增的序列)。

RetryStrategy retry 
  = new RetryStrategy(6, 4, RetryStrategy.BackoffStrategy.EXPONENTIAL_BACKOFF)

writer = (TunnelBufferedWriter) uploadSession.openBufferedWriter();
writer.setRetryStrategy(retry);

缓冲区控制

如果你的程序对 JVM 的内存有严格的要求,可以通过下面这个接口修改缓冲区占内存的字节数(bytes):

writer.setBufferSize(1024*1024);

默认配置每一个 Writer 的 BufferSize 是 10 MiB。TunnelBufferedWriter 一次 flush buffer 的操作上传一个 block 的数据[3]。

多个进程共享 Session

由于一个 Session 的上传状态是通过维护一个 block list 实现的,对于多线程程序来讲,通过锁很容易实现资源的分配。但对于两个进程空间里的程序想要复用一个 Session 时,必须通过一种机制对资源进行隔离。

具体地,在 getUploadSession 的时候,必须指定这个共享这个 Session 的进程数目,以及一个用来区分进程的 global id:

//程序1:这个 session 将被两个 writer 共享,我是其中第 0 个
TableTunnel.UploadSession up 
  = tunnel.getUploadSession(projectName, tableName, sid, 2, 0); 
writer = session.openBufferedWriter();

//程序1:这个 session 将被两个 writer 共享,我是其中第 1 个
TableTunnel.UploadSession up 
  = tunnel.getUploadSession(projectName, tableName, sid, 2, 1); 
writer = session.openBufferedWriter();

Notes

[1] 一次完整的上传流程通常包括以下步骤:

先对数据进行划分
为每个数据块指定 block id,即调用 openRecordWriter(id)
然后用一个或多个线程分别将这些 block 上传上去
并在某个 block 上传失败以后,需要对整个 block 进行重传
在所有 block 都上传以后,向服务端提供上传成功的 blockid list 进行校验,即调用 session.commit([1,2,3,...])
[2] 因为使用长连接,服务端有计时器判断是否客户端是否 alive

[3] block 在服务端有 20000 个的数量上限,如果 BufferSize 设得太小会导致 20000 个 block 很快被用光

[4] Session的有效期为24小时,超过24小时会导致数据上传失败

相关实践学习
基于MaxCompute的热门话题分析
Apsara Clouder大数据专项技能认证配套课程:基于MaxCompute的热门话题分析
目录
相关文章
|
前端开发 Java Shell
【08】flutter完成屏幕适配-重建Android,增加GetX路由,屏幕适配,基础导航栏-多版本SDK以及gradle造成的关于fvm的使用(flutter version manage)-卓伊凡换人优雅草Alex-开发完整的社交APP-前端客户端开发+数据联调|以优雅草商业项目为例做开发-flutter开发-全流程-商业应用级实战开发-优雅草Alex
【08】flutter完成屏幕适配-重建Android,增加GetX路由,屏幕适配,基础导航栏-多版本SDK以及gradle造成的关于fvm的使用(flutter version manage)-卓伊凡换人优雅草Alex-开发完整的社交APP-前端客户端开发+数据联调|以优雅草商业项目为例做开发-flutter开发-全流程-商业应用级实战开发-优雅草Alex
1148 20
【08】flutter完成屏幕适配-重建Android,增加GetX路由,屏幕适配,基础导航栏-多版本SDK以及gradle造成的关于fvm的使用(flutter version manage)-卓伊凡换人优雅草Alex-开发完整的社交APP-前端客户端开发+数据联调|以优雅草商业项目为例做开发-flutter开发-全流程-商业应用级实战开发-优雅草Alex
|
监控 Java 开发工具
【事件中心 Azure Event Hub】Event Hub Java SDK的消费端出现不消费某一个分区中数据的情况,出现IdleTimerExpired错误消息记录
【事件中心 Azure Event Hub】Event Hub Java SDK的消费端出现不消费某一个分区中数据的情况,出现IdleTimerExpired错误消息记录
252 3
|
SQL 存储 分布式计算
"SQLTask携手Tunnel:打造高效海量数据导出解决方案,轻松应对大数据挑战
【8月更文挑战第22天】SQLTask搭配Tunnel实现高效海量数据导出。SQLTask擅长执行复杂查询,但直接导出受限(约1万条)。Tunnel专注数据传输,无大小限制。二者结合,先用SQLTask获取数据,再通过Tunnel高效导出至目标位置(如CSV、OSS等),适用于大数据场景,需配置节点及连接,示例代码展示全过程,满足企业级数据处理需求。
523 2
|
固态存储 Java 网络安全
【Azure Developer】使用Java SDK代码创建Azure VM (包含设置NSG,及添加数据磁盘SSD)
【Azure Developer】使用Java SDK代码创建Azure VM (包含设置NSG,及添加数据磁盘SSD)
349 0
|
JSON Java Serverless
函数计算产品使用问题之如何使用Go SDK从HTTP上下文中提取JSON数据
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。
298 0
|
分布式计算 大数据 Go
MaxCompute操作报错合集之使用go sdk调用GetTunnelEndpoint出现报错:InvalidAction.NotFoundSpecified api is not found,该如何解决
MaxCompute是阿里云提供的大规模离线数据处理服务,用于大数据分析、挖掘和报表生成等场景。在使用MaxCompute进行数据处理时,可能会遇到各种操作报错。以下是一些常见的MaxCompute操作报错及其可能的原因与解决措施的合集。
381 0
|
12月前
|
机器学习/深度学习 传感器 分布式计算
数据才是真救命的:聊聊如何用大数据提升灾难预警的精准度
数据才是真救命的:聊聊如何用大数据提升灾难预警的精准度
726 14
|
数据采集 分布式计算 DataWorks
ODPS在某公共数据项目上的实践
本项目基于公共数据定义及ODPS与DataWorks技术,构建一体化智能化数据平台,涵盖数据目录、归集、治理、共享与开放六大目标。通过十大子系统实现全流程管理,强化数据安全与流通,提升业务效率与决策能力,助力数字化改革。
490 4
|
机器学习/深度学习 运维 监控
运维不怕事多,就怕没数据——用大数据喂饱你的运维策略
运维不怕事多,就怕没数据——用大数据喂饱你的运维策略
1183 0
|
12月前
|
传感器 人工智能 监控
数据下田,庄稼不“瞎种”——聊聊大数据如何帮农业提效
数据下田,庄稼不“瞎种”——聊聊大数据如何帮农业提效
367 14

热门文章

最新文章

相关产品

  • 云原生大数据计算服务 MaxCompute