使用Python，或者使用DataWorks创建PyODPS3节点，读取FTP或者远程FTP文件，然后写入到MaxCompute指定表的分区中。使用的库若未安装，可用pip install方式安装。代码如下：

from odps import ODPS
from odps.df import DataFrame
import numpy as np
import pandas as pd
from odps.df import outputo = ODPS('<odps-key>', 'odps-secret', 'project-name', endpoint='endpoint-address')
t = o.get_table('tmp_xxm_test')my_file = '/Downloads/tmp_xxm_test.csv'field_types = {'账户ID': object,'点击数': np.float64,  # np.float64'花费': object,'时间': object
}
# 先读取数据然后设定类型
data_src = pd.read_csv(my_file, header=0, dtype=field_types)# 修改数据类型，可取代指定每列的类型
# data_src['my_num'] = data_src['my_num'].astype('Int32')data_1 = pd.read_csv('/Downloads/tmp_xxm_test_1.csv', header=0, dtype=field_types)
# 两个DataFrame合并，即UNION
df = pd.concat([data_src, data_1], axis=0, join='outer', ignore_index=False)'''
方式1：自定义方式写入，该方式可以写入到指定分区中，None表示写入null
'''
with t.open_writer(partition='ds=20231128', create_partition=True) as writer:records = data_src.apply(lambda x: [None if pd.isna(x['my_id']) else x['my_id'],None if pd.isna(x['my_num']) else x['my_num'],None if pd.isna(x['my_amt']) else x['my_amt'],None if pd.isna(x['my_time']) else x['my_time']], axis='columns').tolist()print(records)writer.write(records)print('sucess.')'''
方式2：persist方式写入
1. 该方式写入时，目标表不要创建整数字段，因为如果整数字段为np.float的NaN或者pd的NA则无法转为integer。
2. 若data_src中使用'Int64'则创建DataFrame会报错无法识别Int64。
3. 原df的列名在目标表都必须存在，若df中列名缺少目标表的列则该字段填充null。
4. 输出按照列名去输出，不是按照顺序，顺序不一致没关系，但建议一致。
'''
df = DataFrame(data_src)
print(df)
'''
overwrite参数表示是否覆盖，默认True；drop_partition表示是否删除存在的分区数据，默认false。
drop_partition设为False仍然会覆盖已存在的指定分区，overwrite设为False才会Append方式写入。
'''
df.persist('tmp_xxm_test', partition='ds=20231128', drop_partition=True, create_partition=True, odps=o)