通過DataX將Table Store資料匯出到CSV檔案 - Tablestore

背景資訊

DataX是一個異構資料來源離線同步工具，支援MySQL、Oracle、HDFS、Hive、Tablestore等多種資料來源之間的高效穩定同步。

使用DataX匯出資料前，請注意以下事項：

匯出檔案命名：DataX匯出的CSV檔案名稱會附加隨機尾碼（例如 output.csv__d737aec2_c9e3_4489_a5d7_361f44c998ce），匯出完成後需手動移除尾碼以獲得標準CSV檔案名稱。
網路訪問費用：如果在Elastic Compute Service上運行DataX，建議使用VPC地址訪問Table Store執行個體，可避免產生外網下行流量費用，同時獲得更高的網路效能和安全性。

準備工作

擷取Table Store的服務地址、執行個體名稱和源表資訊。
建立阿里雲帳號（主帳號）或RAM使用者的AccessKey。

重要
出於安全考慮，強烈建議通過RAM使用者使用Table Store功能。建立RAM使用者、授予該使用者管理Table Store許可權（AliyunOTSFullAccess）並為該RAM使用者建立AccessKey。具體操作，請參見使用RAM使用者存取金鑰訪問Table Store。

操作步驟

說明

本文以Elastic Compute Service（作業系統：Alibaba Cloud Linux 3.2104 LTS 64位、Ubuntu 22.04 64位）為例介紹具體操作。

步驟一：安裝依賴

安裝Python（Python 2和Python 3均可）。

ECS的Alibaba Cloud Linux和Ubuntu系統已內建Python 3，其他系統請自行安裝。
安裝JDK（1.8及以上，推薦1.8）。

以下為在ECS的Alibaba Cloud Linux或Ubuntu系統中安裝JDK 1.8的命令，其他系統請自行安裝。
Alibaba Cloud Linux
```
yum -y install java-1.8.0-openjdk-devel.x86_64
```
Ubuntu
```
apt update && apt upgrade
apt install openjdk-8-jdk
```

步驟二：下載DataX

下載DataX工具包。

wget https://datax-opensource.oss-cn-hangzhou.aliyuncs.com/202309/datax.tar.gz

解壓工具包。
```
tar -zxvf datax.tar.gz
```

如需自行編譯DataX，請參見DataX安裝指引。

步驟三：編寫設定檔

進入DataX的bin目錄。
```
cd datax/bin
```

建立設定檔。如果使用的是vim，請自行替換命令。

vi tablestore_to_csv.json

以下為不同表類型的設定檔樣本，請根據實際情況調整相關參數。

匯出資料表

資料表源表 orders 的結構資訊，請參見附錄1：資料表。

{
  "job": {
    "setting": {
      "speed": {
        "channel": 1
      },
      "errorLimit": {
        "record": 0,
        "percentage": 0
      }
    },
    "content": [
      {
        "reader": {
          "name": "otsreader",
          "parameter": {
            "endpoint": "https://<YOUR-INSTANCE>.<YOUR-REGION>.ots.aliyuncs.com",
            "accessId": "<YOUR-ACCESS-KEY-ID>",
            "accessKey": "<YOUR-ACCESS-KEY-SECRET>",
            "instanceName": "<YOUR-INSTANCE>",
            "table": "orders",
            "range": {
              "begin": [
                {
                  "type": "INF_MIN"
                }
              ],
              "end": [
                {
                  "type": "INF_MAX"
                }
              ]
            },
            "column": [
              {
                "name": "order_id"
              },
              {
                "name": "user_id"
              },
              {
                "name": "sku_id"
              },
              {
                "name": "price"
              },
              {
                "name": "num"
              },
              {
                "name": "total_price"
              },
              {
                "name": "order_status"
              },
              {
                "name": "create_time"
              },
              {
                "name": "modified_time"
              }
            ]
          }
        },
        "writer": {
          "name": "txtfilewriter",
          "parameter": {
            "path": "/tmp/export/",
            "fileName": "output.csv",
            "writeMode": "truncate",
            "fileFormat": "csv"
          }
        }
      }
    ]
  }
}

匯出時序表

時序表源表 vehicles_timeseriesdata 的結構資訊，請參見附錄2：時序表。

{
  "job": {
    "setting": {
      "speed": {
        "channel": 1
      }
    },
    "content": [
      {
        "reader": {
          "name": "otsreader",
          "parameter": {
            "endpoint": "https://<YOUR-INSTANCE>.<YOUR-REGION>.ots.aliyuncs.com",
            "accessId": "<YOUR-ACCESS-KEY-ID>",
            "accessKey": "<YOUR-ACCESS-KEY-SECRET>",
            "instanceName": "<YOUR-INSTANCE>",
            "table": "vehicles_timeseriesdata",
            "mode": "normal",
            "newVersion": "true",
            "isTimeseriesTable": "true",
            "measurementName": "measurement_1",
            "timeRange": {
              "begin": 0,
              "end": 1750000000000
            },
            "column": [
              {
                "name": "_m_name"
              },
              {
                "name": "_data_source"
              },
              {
                "name": "_tags"
              },
              {
                "name": "_time"
              },
              {
                "name": "vin_id",
                "type": "STRING"
              },
              {
                "name": "mfrs",
                "type": "STRING"
              },
              {
                "name": "model",
                "type": "STRING"
              },
              {
                "name": "speed",
                "type": "DOUBLE"
              },
              {
                "name": "gps",
                "type": "STRING"
              },
              {
                "name": "mileage",
                "type": "DOUBLE"
              },
              {
                "name": "emission",
                "type": "DOUBLE"
              }
            ]
          }
        },
        "writer": {
          "name": "txtfilewriter",
          "parameter": {
            "path": "/tmp/export/",
            "fileName": "output.csv",
            "writeMode": "truncate",
            "fileFormat": "csv"
          }
        }
      }
    ]
  }
}

OTSReader需要替換的參數說明如下：

參數名稱	適用表類型	說明
channel	兩者	資料同步任務的並發通道數。每個channel對應一個獨立的讀寫線程，調整此值可控制任務並發度，進而影響系統效能和資源消耗。
endpoint	兩者	Tablestore執行個體的服務地址。通過Elastic Compute Service訪問Tablestore時，建議使用VPC地址，可避免外網下行流量費用，並提供更高的網路效能和安全性。
accessId	兩者	阿里雲帳號或RAM使用者的AccessKey ID。
accessKey	兩者	阿里雲帳號或RAM使用者的AccessKey Secret。
instanceName	兩者	Tablestore執行個體的名稱。
tableName	兩者	Tablestore的源表名稱。
column	兩者	需要匯出的列數組。
range	資料表	資料讀取的主鍵範圍，begin和end分別為起始主鍵和結束主鍵，格式為JSON數組，讀取範圍為左閉右開區間。起始主鍵和結束主鍵需要是有效主鍵或由`INF_MIN`（無限小）和`INF_MAX`（無限大）類型組成的虛擬點，虛擬點的列數必須與主鍵相同。此參數為可選項，預設從無限小開始讀取。
measurementName	時序表	需要讀取時間軸的度量名稱。未配置時讀取全表資料。
timeRange	時序表	資料讀取的時間戳記範圍，begin和end分別為起始時間戳記和結束時間戳記，單位為毫秒，讀取範圍為左閉右開區間。此參數為可選項，預設讀取全部版本。

TxtFileWriter需要替換的參數說明如下：

參數名稱	說明
path	匯出檔案在系統中的儲存路徑。
fileName	匯出檔案的名稱，需包含副檔名，例如`output.csv`。
writeMode	寫入資料前的檔案處理模式，取值如下： `truncate`：寫入前刪除目標目錄下所有以fileName為首碼的檔案。 `append`：直接寫入檔案，不做預先處理，並保證檔案名稱不衝突。 `nonConflict`：目標目錄下已存在以fileName為首碼的檔案時，報錯並中止操作。
fileFormat	檔案寫出的格式，支援`csv`和`text`兩種類型。

步驟四：執行DataX任務

執行以下命令開始匯出資料。

python3 datax.py tablestore_to_csv.json

任務結束後，將列印整體運行情況，樣本如下：

2025-03-19 17:21:05.146 [job-0] INFO  StandAloneJobContainerCommunicator - Total 200000 records, 23086634 bytes | Speed 1.10MB/s, 10000 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.222s |  All Task WaitReaderTime 18.392s | Percentage 100.00%
2025-03-19 17:21:05.147 [job-0] INFO  JobContainer - 
任務啟動時刻                    : 2025-03-19 17:20:43
任務結束時刻                    : 2025-03-19 17:21:05
任務總計耗時                    :                 21s
任務平均流量                    :            1.10MB/s
記錄寫入速度                    :          10000rec/s
讀出記錄總數                    :              200000
讀寫失敗總數                    :                   0

驗證匯出結果。

任務執行完成後，檢查指定路徑（如/tmp/export/），確認匯出的CSV檔案是否符合預期。匯出檔案名稱包含隨機尾碼，樣本如下：
```
output.csv__d737aec2_c9e3_4489_a5d7_361f44c998ce
```
預覽檔案內容，驗證資料是否正確：
```
head -5 output.csv__d737aec2_c9e3_4489_a5d7_361f44c998ce
```
輸出樣本如下：
```
order_id,user_id,sku_id,price,num,total_price,order_status,create_time,modified_time
ORD-001,USR-100,SKU-200,29.99,2,59.98,completed,2025-01-01 08:00:00,2025-01-02 10:00:00
ORD-002,USR-101,SKU-201,49.99,1,49.99,pending,2025-01-02 09:00:00,2025-01-02 09:00:00
```
移除隨機尾碼以獲得標準CSV檔案名稱：
```
mv output.csv__d737aec2_c9e3_4489_a5d7_361f44c998ce output.csv
```
說明
匯出的CSV檔案名稱將附加隨機尾碼，用於區分多線程並發寫入的檔案。需手動移除此尾碼以獲得標準的CSV檔案。

附錄-源表範例

附錄1：資料表

資料表範例的名稱為orders，表結構資訊請參見下表。

欄位名稱	類型	描述
order_id（主鍵）	String	訂單ID。
user_id	String	使用者ID。
sku_id	String	商品ID。
price	Double	商品購買單價。
num	Integer	商品購買數量。
total_price	Double	訂單總價。
order_status	String	訂單狀態。
create_time	String	訂單建立時間。
modified_time	String	最後修改時間。

附錄2：時序表

時序表範例的名稱為vehicles_timeseriesdata，表結構資訊請參見下表。

欄位名稱	類型	描述
_m_name	String	度量名稱，表示時間軸資料所度量的物理量或者監控指標的名稱。
_data_source	String	資料來源，表示產生時間軸的資料來源標識，可以為空白。
_tags	String	時間軸的標籤資訊。
_time	Integer	資料上報時間。
vin_id	String	車輛識別代碼，即車架號，用於唯一標識車輛。
mfrs	String	生產廠商。
model	String	車輛型號。
speed	Double	當前速度。
gps	String	車輛GPS座標，格式為`緯度,經度`，緯度在前，經度在後。
mileage	Double	當前裡程數。
emission	Double	排放值。