阿里云MaxCompute云原生大数据计算服务全方位对接使用指南

# AccessKey信息
access_id=你的AccessKey ID
access_key=你的AccessKey Secret
# 目标项目名称
project_name=my_project
# MaxCompute服务Endpoint
end_point=https://service.cn-hangzhou.maxcompute.aliyun.com/api
# Tunnel服务Endpoint
tunnel_endpoint=https://dt.cn-hangzhou.maxcompute.aliyun.com
# LogView地址
log_view_host=http://logview.odps.aliyun.com

-- 查看当前项目中的所有表
show tables;
-- 创建分区表
CREATE TABLE IF NOT EXISTS sales (
    order_id STRING,
    product_name STRING,
    amount DOUBLE
) PARTITIONED BY (dt STRING);
-- 通过Tunnel命令上传本地数据
Tunnel upload /path/to/data.csv sales PARTITION(dt='2026-06-18') -c 'UTF-8';
-- 执行SQL查询
SELECT dt, COUNT(*) AS order_count, SUM(amount) AS total_amount
FROM sales
WHERE dt >= '2026-06-01'
GROUP BY dt
ORDER BY dt DESC;
-- 查看作业的LogView
WAIT .;  -- 在SQL语句后使用分号提交，系统会返回Instance ID和LogView URL

{
  "type": "job",
  "steps": [
    {
      "stepType": "mysql",
      "parameter": {
        "datasource": "mysql_datasource",
        "column": ["id", "name", "created_at"],
        "table": "source_table",
        "where": "created_at >= '2026-01-01'"
      },
      "name": "Reader",
      "category": "reader"
    },
    {
      "stepType": "odps",
      "parameter": {
        "datasource": "odps_datasource",
        "column": ["id", "name", "created_at"],
        "table": "target_table",
        "partition": "dt=${bizdate}"
      },
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
    "errorLimit": {"record": 0},
    "speed": {"concurrent": 2}
  }
}

<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>odps-sdk-core</artifactId>
    <version>0.27.2-public</version>
</dependency>

import com.aliyun.odps.Odps;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.Account;
// 从环境变量读取AccessKey，避免硬编码
String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
Account account = new AliyunAccount(accessId, accessKey);
Odps odps = new Odps(account);
// 设置Endpoint和默认项目
odps.setEndpoint("https://service.cn-hangzhou.maxcompute.aliyun.com/api");
odps.setDefaultProject("my_project");
// 遍历项目中的所有表
for (Table t : odps.tables()) {
    System.out.println("Table: " + t.getName());
}

import com.aliyun.odps.tunnel.TunnelClient;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.tunnel.UploadSession;
import com.aliyun.odps.data.Record;
// 创建Tunnel客户端
TunnelClient tunnelClient = new TunnelClient(odps);
TableTunnel tunnel = new TableTunnel(tunnelClient);
// 创建上传会话
UploadSession uploadSession = tunnel.createUploadSession("my_project", "sales");
// 准备数据并写入
for (int i = 0; i < 10000; i++) {
    Record record = uploadSession.newRecord();
    record.setString("order_id", "ORD" + i);
    record.setString("product_name", "Product_" + (i % 100));
    record.setDouble("amount", Math.random() * 1000);
    uploadSession.writeRecord(record);
}
// 提交上传会话
uploadSession.commit();
System.out.println("Upload completed. Block count: " + uploadSession.getBlockCount());

pip install pyodps

from odps import ODPS
import os
# 从环境变量读取AccessKey
access_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')
access_key = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
project = 'my_project'
endpoint = 'https://service.cn-hangzhou.maxcompute.aliyun.com/api'
o = ODPS(access_id, access_key, project, endpoint)

# 执行SQL查询并获取结果
with o.execute_sql('SELECT dt, COUNT(*) as cnt FROM sales GROUP BY dt').open_reader() as reader:
    for record in reader:
        print(record[0], record[1])
# 使用DataFrame API进行数据处理
from odps import DataFrame
# 将MaxCompute表转换为DataFrame
df = DataFrame(o.get_table('sales'))
# 进行过滤、聚合等操作
result = df[df.amount > 100].groupby('dt').agg(count=df.order_id.count(), total=df.amount.sum())
# 执行计算并收集结果
for record in result.head(10):
    print(record)
# 创建表
o.create_table('new_table', {'id': 'string', 'value': 'double'}, if_not_exists=True)
# 将本地数据写入MaxCompute表
from odps.df import to_pandas
# 读取表数据到Pandas DataFrame（适用于小数据集）
pdf = o.get_table('sales').to_pandas()
# 处理后写回
from odps import options
options.df.quote = False  # 关闭自动转义
df_new = DataFrame(pdf)
o.write_table('sales_processed', df_new, partition='dt=2026-06-18')

<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>odps-jdbc</artifactId>
    <version>3.9.0</version>
</dependency>

jdbc:odps:https://service.cn-hangzhou.maxcompute.aliyun.com/api?project=my_project

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
public class MaxComputeJDBCDemo {
    public static void main(String[] args) throws Exception {
        String url = "jdbc:odps:https://service.cn-hangzhou.maxcompute.aliyun.com/api?project=my_project";
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        
        Connection conn = DriverManager.getConnection(url, accessId, accessKey);
        Statement stmt = conn.createStatement();
        ResultSet rs = stmt.executeQuery("SELECT dt, COUNT(*) FROM sales GROUP BY dt");
        
        while (rs.next()) {
            System.out.println(rs.getString(1) + ": " + rs.getLong(2));
        }
        
        rs.close();
        stmt.close();
        conn.close();
    }
}

import com.aliyun.odps.Odps;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.table.api.*;
public class StorageAPIDemo {
    public static void main(String[] args) {
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        String project = "my_project";
        String endpoint = "https://service.cn-hangzhou.maxcompute.aliyun.com/api";
        String quotaName = "pay-as-you-go";  // 按量付费资源组名称
        
        Account account = new AliyunAccount(accessId, accessKey);
        Odps odps = new Odps(account);
        odps.setDefaultProject(project);
        odps.setEndpoint(endpoint);
        
        // 配置凭证和环境
        Credentials credentials = Credentials.newBuilder()
                .withAccount(odps.getAccount())
                .withAppAccount(odps.getAppAccount())
                .build();
        EnvironmentSettings settings = EnvironmentSettings.newBuilder()
                .withCredentials(credentials)
                .withServiceEndpoint(odps.getEndpoint())
                .withQuotaName(quotaName)
                .build();
        
        // 创建读会话
        TableReadSessionBuilder builder = TableReadSessionBuilder.newBuilder()
                .withTableName("sales")
                .withSettings(settings);
        TableBatchReadSession session = builder.build();
        
        // 获取数据分片并读取
        for (Split split : session.getSplits()) {
            try (SplitReader reader = session.createReader(split)) {
                while (reader.hasNext()) {
                    Row row = reader.next();
                    System.out.println(row.getString(0) + ": " + row.getDouble(2));
                }
            }
        }
    }
}

CREATE EXTERNAL TABLE IF NOT EXISTS ots_vehicle_track (
    vid STRING,
    gt BIGINT,
    location STRING
)
STORED BY 'com.aliyun.odps.TablestoreStorageHandler'
WITH SERDEPROPERTIES (
    'tablestore.instance.name'='cap1',
    'tablestore.table.name'='vehicle_track',
    'tablestore.access.id'='${access_id}',
    'tablestore.access.key'='${access_key}',
    'tablestore.endpoint'='https://cap1.cn-hangzhou.ots-internal.aliyuncs.com'
);

SELECT vid, COUNT(*) FROM ots_vehicle_track GROUP BY vid;

{
  "Version": "1",
  "Statement": [
    {
      "Action": [
        "odps:Select",
        "odps:Describe"
      ],
      "Resource": [
        "acs:odps:*:projects/my_project/tables/sales"
      ],
      "Effect": "Allow"
    }
  ]
}