• DataX二次开发——（9）新增s3reader和s3writer模块

    1 背景

    DataX3.0支持阿里的OSS的读写，但没支持S3的读写，虽然OSS的也是基于S3协议去做二开的，但是一些参数有点区别，所以参考阿里的OSSReader和OSSWriter的设计开发了S3Reader和S3Writer。

    2 代码开发

    2.1 s3reader

    2.1.1 项目结构

     2.1.2 代码如下

    package.xml

    1. <assembly
    2.         xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
    3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.         xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
    5.     <formats>
    6.         <format>dir</format>
    7.     </formats>
    8.     <includeBaseDirectory>false</includeBaseDirectory>
    9.     <fileSets>
    10.         <fileSet>
    11.             <directory>src/main/resources</directory>
    12.             <includes>
    13.                 <include>plugin.json</include>
    14.                 <include>plugin_job_template.json</include>
    15.             </includes>
    16.             <outputDirectory>plugin/reader/s3reader</outputDirectory>
    17.         </fileSet>
    18.         <fileSet>
    19.             <directory>target/</directory>
    20.             <includes>
    21.                 <include>s3reader-0.0.1-SNAPSHOT.jar</include>
    22.             </includes>
    23.             <outputDirectory>plugin/reader/s3reader</outputDirectory>
    24.         </fileSet>
    25.     </fileSets>
    26.  
    27.     <dependencySets>
    28.         <dependencySet>
    29.             <useProjectArtifact>false</useProjectArtifact>
    30.             <outputDirectory>plugin/reader/s3reader/libs</outputDirectory>
    31.             <scope>runtime</scope>
    32.         </dependencySet>
    33.     </dependencySets>
    34. </assembly>

    Constant.java

    1. package com.alibaba.datax.plugin.reader.s3reader;
    2.  
    3. public class Constant {
    4.     public static final String OBJECT = "object";
    5.     public static final int CONNECT_TIMEOUT = 50000;
    6.     public static final String SIGNER_OVERRIDE = "S3SignerType";
    7. }

     Key.java

    1. public class Key {
    2.     public static final String ENDPOINT = "endpoint";
    3.  
    4.     public static final String ACCESSKEY = "accessKey";
    5.  
    6.     public static final String SECRETKEY = "secretKey";
    7.  
    8.     public static final String BUCKET = "bucket";
    9.  
    10.     public static final String OBJECT = "object";
    11. }

    S3Reader.java

    1. package com.alibaba.datax.plugin.reader.s3reader;
    2.  
    3. import com.alibaba.datax.common.exception.DataXException;
    4. import com.alibaba.datax.common.plugin.RecordSender;
    5. import com.alibaba.datax.common.spi.Reader;
    6. import com.alibaba.datax.common.util.Configuration;
    7. import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderUtil;
    8. import com.amazonaws.services.s3.AmazonS3;
    9. import com.amazonaws.services.s3.model.ListObjectsRequest;
    10. import com.amazonaws.services.s3.model.ObjectListing;
    11. import com.amazonaws.services.s3.model.S3Object;
    12. import com.amazonaws.services.s3.model.S3ObjectSummary;
    13. import com.google.common.collect.Sets;
    14. import org.apache.commons.io.Charsets;
    15. import org.apache.commons.io.IOUtils;
    16. import org.apache.commons.lang3.StringUtils;
    17. import org.slf4j.Logger;
    18. import org.slf4j.LoggerFactory;
    19.  
    20. import java.io.InputStream;
    21. import java.nio.charset.UnsupportedCharsetException;
    22. import java.util.ArrayList;
    23. import java.util.List;
    24. import java.util.Set;
    25. import java.util.regex.Pattern;
    26.  
    27.  
    28. public class S3Reader extends Reader {
    29.     public static class Job extends Reader.Job {
    30.         private static final Logger LOG = LoggerFactory.getLogger(Job.class);
    31.  
    32.         private Configuration readerOriginConfig = null;
    33.  
    34.         @Override
    35.         public void init() {
    36.             LOG.debug("init() begin...");
    37.             this.readerOriginConfig = this.getPluginJobConf();
    38.             this.validate();
    39.             LOG.debug("init() ok and end...");
    40.         }
    41.  
    42.         private void validate() {
    43.             String endpoint = this.readerOriginConfig.getString(Key.ENDPOINT);
    44.             if (StringUtils.isBlank(endpoint)) {
    45.                 throw DataXException.asDataXException(
    46.                         S3ReaderErrorCode.CONFIG_INVALID_EXCEPTION,
    47.                         "您需要指定 endpoint");
    48.             }
    49.  
    50.             String secretKey = this.readerOriginConfig.getString(Key.SECRETKEY);
    51.             if (StringUtils.isBlank(secretKey)) {
    52.                 throw DataXException.asDataXException(
    53.                         S3ReaderErrorCode.CONFIG_INVALID_EXCEPTION,
    54.                         "您需要指定 secretKey");
    55.             }
    56.  
    57.             String accessKey = this.readerOriginConfig.getString(Key.ACCESSKEY);
    58.             if (StringUtils.isBlank(accessKey)) {
    59.                 throw DataXException.asDataXException(
    60.                         S3ReaderErrorCode.CONFIG_INVALID_EXCEPTION,
    61.                         "您需要指定 accessKey");
    62.             }
    63.  
    64.             String bucket = this.readerOriginConfig.getString(Key.BUCKET);
    65.             if (StringUtils.isBlank(bucket)) {
    66.                 throw DataXException.asDataXException(
    67.                         S3ReaderErrorCode.CONFIG_INVALID_EXCEPTION,
    68.                         "您需要指定 bucket");
    69.             }
    70.  
    71.             String object = this.readerOriginConfig.getString(Key.OBJECT);
    72.             if (StringUtils.isBlank(object)) {
    73.                 throw DataXException.asDataXException(
    74.                         S3ReaderErrorCode.CONFIG_INVALID_EXCEPTION,
    75.                         "您需要指定 object");
    76.             }
    77.  
    78.             String fieldDelimiter = this.readerOriginConfig
    79.                     .getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.FIELD_DELIMITER);
    80.             // warn: need length 1
    81.             if (null == fieldDelimiter || fieldDelimiter.length() == 0) {
    82.                 throw DataXException.asDataXException(
    83.                         S3ReaderErrorCode.CONFIG_INVALID_EXCEPTION,
    84.                         "您需要指定 fieldDelimiter");
    85.             }
    86.  
    87.             String encoding = this.readerOriginConfig
    88.                     .getString(
    89.                             com.alibaba.datax.plugin.unstructuredstorage.reader.Key.ENCODING,
    90.                             com.alibaba.datax.plugin.unstructuredstorage.reader.Constant.DEFAULT_ENCODING);
    91.             try {
    92.                 Charsets.toCharset(encoding);
    93.             } catch (UnsupportedCharsetException uce) {
    94.                 throw DataXException.asDataXException(
    95.                         S3ReaderErrorCode.ILLEGAL_VALUE,
    96.                         String.format("不支持的编码格式 : [%s]", encoding), uce);
    97.             } catch (Exception e) {
    98.                 throw DataXException.asDataXException(
    99.                         S3ReaderErrorCode.ILLEGAL_VALUE,
    100.                         String.format("运行配置异常 : %s", e.getMessage()), e);
    101.             }
    102.  
    103.             // 检测是column 是否为 ["*"] 若是则填为空
    104.             List<Configuration> column = this.readerOriginConfig
    105.                     .getListConfiguration(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN);
    106.             if (null != column
    107.                     && 1 == column.size()
    108.                     && ("\"*\"".equals(column.get(0).toString()) || "'*'"
    109.                     .equals(column.get(0).toString()))) {
    110.                 readerOriginConfig
    111.                         .set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN,
    112.                                 new ArrayList<String>());
    113.             } else {
    114.                 // column: 1. index type 2.value type 3.when type is Data, may
    115.                 // have
    116.                 // format
    117.                 List<Configuration> columns = this.readerOriginConfig
    118.                         .getListConfiguration(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN);
    119.  
    120.                 if (null == columns || columns.size() == 0) {
    121.                     throw DataXException.asDataXException(
    122.                             S3ReaderErrorCode.CONFIG_INVALID_EXCEPTION,
    123.                             "您需要指定 columns");
    124.                 }
    125.  
    126.                 if (null != columns && columns.size() != 0) {
    127.                     for (Configuration eachColumnConf : columns) {
    128.                         eachColumnConf
    129.                                 .getNecessaryValue(
    130.                                         com.alibaba.datax.plugin.unstructuredstorage.reader.Key.TYPE,
    131.                                         S3ReaderErrorCode.REQUIRED_VALUE);
    132.                         Integer columnIndex = eachColumnConf
    133.                                 .getInt(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.INDEX);
    134.                         String columnValue = eachColumnConf
    135.                                 .getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.VALUE);
    136.  
    137.                         if (null == columnIndex && null == columnValue) {
    138.                             throw DataXException.asDataXException(
    139.                                     S3ReaderErrorCode.NO_INDEX_VALUE,
    140.                                     "由于您配置了type, 则至少需要配置 index 或 value");
    141.                         }
    142.  
    143.                         if (null != columnIndex && null != columnValue) {
    144.                             throw DataXException.asDataXException(
    145.                                     S3ReaderErrorCode.MIXED_INDEX_VALUE,
    146.                                     "您混合配置了index, value, 每一列同时仅能选择其中一种");
    147.                         }
    148.  
    149.                     }
    150.                 }
    151.             }
    152.  
    153.             // only support compress: gzip,bzip2,zip
    154.             String compress = this.readerOriginConfig.getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COMPRESS);
    155.             if (StringUtils.isBlank(compress)) {
    156.                 this.readerOriginConfig.set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COMPRESS, null);
    157.             } else {
    158.                 Set<String> supportedCompress = Sets.newHashSet("gzip", "bzip2", "zip");
    159.                 compress = compress.toLowerCase().trim();
    160.                 if (!supportedCompress.contains(compress)) {
    161.                     throw DataXException.asDataXException(
    162.                             S3ReaderErrorCode.ILLEGAL_VALUE,
    163.                             String.format(
    164.                                     "仅支持 gzip, bzip2, zip 文件压缩格式 , 不支持您配置的文件压缩格式: [%s]",
    165.                                     compress));
    166.                 }
    167.                 this.readerOriginConfig
    168.                         .set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COMPRESS,
    169.                                 compress);
    170.             }
    171.         }
    172.  
    173.         @Override
    174.         public void prepare() {
    175.             LOG.debug("prepare()");
    176.         }
    177.  
    178.         @Override
    179.         public void post() {
    180.             LOG.debug("post()");
    181.         }
    182.  
    183.         @Override
    184.         public void destroy() {
    185.             LOG.debug("destroy()");
    186.         }
    187.  
    188.         @Override
    189.         public List<Configuration> split(int adviceNumber) {
    190.             LOG.debug("split() begin...");
    191.             List<Configuration> readerSplitConfigs = new ArrayList<Configuration>();
    192.  
    193.             // 将每个单独的 object 作为一个 slice
    194.             List<String> objects = parseOriginObjects(readerOriginConfig
    195.                     .getList(Constant.OBJECT, String.class));
    196.             if (0 == objects.size()) {
    197.                 throw DataXException.asDataXException(
    198.                         S3ReaderErrorCode.EMPTY_BUCKET_EXCEPTION,
    199.                         String.format(
    200.                                 "未能找到待读取的Object,请确认您的配置项bucket: %s object: %s",
    201.                                 this.readerOriginConfig.get(Key.BUCKET),
    202.                                 this.readerOriginConfig.get(Key.OBJECT)));
    203.             }
    204.  
    205.             for (String object : objects) {
    206.                 Configuration splitedConfig = this.readerOriginConfig.clone();
    207.                 splitedConfig.set(Constant.OBJECT, object);
    208.                 readerSplitConfigs.add(splitedConfig);
    209.                 LOG.info(String.format("S3 object to be read:%s", object));
    210.             }
    211.             LOG.debug("split() ok and end...");
    212.             return readerSplitConfigs;
    213.         }
    214.  
    215.         private List<String> parseOriginObjects(List<String> originObjects) {
    216.             List<String> parsedObjects = new ArrayList<String>();
    217.  
    218.             for (String object : originObjects) {
    219.                 int firstMetaChar = (object.indexOf('*') > object.indexOf('?')) ? object
    220.                         .indexOf('*') : object.indexOf('?');
    221.  
    222.                 if (firstMetaChar != -1) {
    223.                     int lastDirSeparator = object.lastIndexOf(
    224.                             IOUtils.DIR_SEPARATOR, firstMetaChar);
    225.                     String parentDir = object
    226.                             .substring(0, lastDirSeparator + 1);
    227.                     List<String> remoteObjects = getRemoteObjects(parentDir);
    228.                     Pattern pattern = Pattern.compile(object.replace("*", ".*")
    229.                             .replace("?", ".?"));
    230.  
    231.                     for (String remoteObject : remoteObjects) {
    232.                         if (pattern.matcher(remoteObject).matches()) {
    233.                             parsedObjects.add(remoteObject);
    234.                         }
    235.                     }
    236.                 } else {
    237.                     parsedObjects.add(object);
    238.                 }
    239.             }
    240.             return parsedObjects;
    241.         }
    242.  
    243.         private List<String> getRemoteObjects(String parentDir) {
    244.  
    245.             LOG.debug(String.format("父文件夹 : %s", parentDir));
    246.             List<String> remoteObjects = new ArrayList<String>();
    247.             AmazonS3 client = S3Util.initS3Client(readerOriginConfig);
    248.             try {
    249.                 ListObjectsRequest listObjectsRequest = new ListObjectsRequest(
    250.                         readerOriginConfig.getString(Key.BUCKET), parentDir, null, null, null
    251.                 );
    252.                 ObjectListing objectList;
    253.                 do {
    254.                     objectList = client.listObjects(listObjectsRequest);
    255.                     for (S3ObjectSummary objectSummary : objectList
    256.                             .getObjectSummaries()) {
    257.                         LOG.debug(String.format("找到文件 : %s",
    258.                                 objectSummary.getKey()));
    259.                         remoteObjects.add(objectSummary.getKey());
    260.                     }
    261.                     listObjectsRequest.setMarker(objectList.getNextMarker());
    262.                     LOG.debug(listObjectsRequest.getMarker());
    263.                     LOG.debug(String.valueOf(objectList.isTruncated()));
    264.  
    265.                 } while (objectList.isTruncated());
    266.             } catch (IllegalArgumentException e) {
    267.                 throw DataXException.asDataXException(
    268.                         S3ReaderErrorCode.S3_EXCEPTION, e.getMessage());
    269.             }
    270.  
    271.             return remoteObjects;
    272.         }
    273.     }
    274.  
    275.     public static class Task extends Reader.Task {
    276.         private static Logger LOG = LoggerFactory.getLogger(Reader.Task.class);
    277.  
    278.         private Configuration readerSliceConfig;
    279.  
    280.         @Override
    281.         public void startRead(RecordSender recordSender) {
    282.             LOG.debug("read start");
    283.             String bucket = readerSliceConfig.getString(Key.BUCKET);
    284.             String object = readerSliceConfig.getString(Key.OBJECT);
    285.             AmazonS3 client = S3Util.initS3Client(readerSliceConfig);
    286.  
    287.             S3Object s3Object = client.getObject(bucket, object);
    288.             InputStream objectStream = s3Object.getObjectContent();
    289.             UnstructuredStorageReaderUtil.readFromStream(objectStream, object,
    290.                     this.readerSliceConfig, recordSender,
    291.                     this.getTaskPluginCollector());
    292.             recordSender.flush();
    293.         }
    294.  
    295.         @Override
    296.         public void init() {
    297.             this.readerSliceConfig = this.getPluginJobConf();
    298.         }
    299.  
    300.         @Override
    301.         public void destroy() {
    302.  
    303.         }
    304.     }
    305. }

    S3ReaderErrorCode.java

    1. package com.alibaba.datax.plugin.reader.s3reader;
    2.  
    3. import com.alibaba.datax.common.spi.ErrorCode;
    4.  
    5. public enum S3ReaderErrorCode implements ErrorCode {
    6.     // TODO: 修改错误码类型
    7.     RUNTIME_EXCEPTION("S3Reader-00", "运行时异常"),
    8.     S3_EXCEPTION("OssFileReader-01", "OSS配置异常"),
    9.     CONFIG_INVALID_EXCEPTION("OssFileReader-02", "参数配置错误"),
    10.     NOT_SUPPORT_TYPE("S3Reader-03", "不支持的类型"),
    11.     CAST_VALUE_TYPE_ERROR("OssFileReader-04", "无法完成指定类型的转换"),
    12.     SECURITY_EXCEPTION("S3Reader-05", "缺少权限"),
    13.     ILLEGAL_VALUE("S3Reader-06", "值错误"),
    14.     REQUIRED_VALUE("S3Reader-07", "必选项"),
    15.     NO_INDEX_VALUE("S3Reader-08", "没有 Index"),
    16.     MIXED_INDEX_VALUE("S3Reader-09", "index 和 value 混合"),
    17.     EMPTY_BUCKET_EXCEPTION("S3Reader-10", "您尝试读取的Bucket为空");
    18.  
    19.     private final String code;
    20.     private final String description;
    21.  
    22.     private S3ReaderErrorCode(String code, String description) {
    23.         this.code = code;
    24.         this.description = description;
    25.     }
    26.  
    27.     @Override
    28.     public String getCode() {
    29.         return this.code;
    30.     }
    31.  
    32.     @Override
    33.     public String getDescription() {
    34.         return this.description;
    35.     }
    36.  
    37.     @Override
    38.     public String toString() {
    39.         return String.format("Code:[%s], Description:[%s].", this.code,
    40.                 this.description);
    41.     }
    42. }

     S3Util.java

    1. package com.alibaba.datax.plugin.reader.s3reader;
    2.  
    3. import com.alibaba.datax.common.util.Configuration;
    4. import com.amazonaws.ClientConfiguration;
    5. import com.amazonaws.auth.AWSCredentials;
    6. import com.amazonaws.auth.AWSCredentialsProvider;
    7. import com.amazonaws.auth.AWSStaticCredentialsProvider;
    8. import com.amazonaws.auth.BasicAWSCredentials;
    9. import com.amazonaws.client.builder.AwsClientBuilder;
    10. import com.amazonaws.services.s3.AmazonS3;
    11. import com.amazonaws.services.s3.AmazonS3ClientBuilder;
    12.  
    13. import java.util.HashMap;
    14. import java.util.Map;
    15.  
    16. /**
    17.  * @Description: obs 工具类
    18.  * @Author: chenweifeng
    19.  * @Date: 2022年11月18日 下午1:56
    20.  **/
    21. public class S3Util {
    22.     public static void main(String[] args) {
    23.         Map<String, Object> map = new HashMap<>();
    24.         map.put(Key.ENDPOINT, "http://x.x.x.x:9383");
    25.         map.put(Key.ACCESSKEY, "admin");
    26.         map.put(Key.SECRETKEY, "123456");
    27.         Configuration conf = Configuration.from(map);
    28.         initS3Client(conf);
    29.     }
    30.  
    31.     /**
    32.      * 初始化S3Client
    33.      *
    34.      * @param conf
    35.      * @return
    36.      */
    37.     public static AmazonS3 initS3Client(Configuration conf) {
    38.         String endpoint = conf.getString(Key.ENDPOINT);
    39.         String accessKey = conf.getString(Key.ACCESSKEY);
    40.         String secretKey = conf.getString(Key.SECRETKEY);
    41.  
    42.         AmazonS3ClientBuilder amazonS3ClientBuilder = AmazonS3ClientBuilder.standard();
    43.  
    44.         ClientConfiguration clientConfiguration = new ClientConfiguration();
    45.         clientConfiguration.setProtocol(com.amazonaws.Protocol.HTTPS);
    46.         clientConfiguration.setConnectionTimeout(Constant.CONNECT_TIMEOUT);
    47.         clientConfiguration.setSignerOverride(Constant.SIGNER_OVERRIDE);
    48.         amazonS3ClientBuilder.setClientConfiguration(clientConfiguration);
    49.  
    50.         AWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
    51.         AWSCredentialsProvider awsCredentialsProvider = new AWSStaticCredentialsProvider(awsCredentials);
    52.         amazonS3ClientBuilder.setCredentials(awsCredentialsProvider);
    53.  
    54.         AwsClientBuilder.EndpointConfiguration endpointConfiguration = new AwsClientBuilder.EndpointConfiguration(endpoint, null);
    55.         amazonS3ClientBuilder.setEndpointConfiguration(endpointConfiguration);
    56.  
    57.         AmazonS3 amazonS3 = amazonS3ClientBuilder.build();
    58.         return amazonS3;
    59.     }
    60. }

    plugin.json

    1. {
    2.     "name": "s3reader",
    3.     "class": "com.alibaba.datax.plugin.reader.s3reader.S3Reader",
    4.     "description": "",
    5.     "developer": "alibaba"
    6. }

    2.2 s3writer

    2.2.1 项目结构

     2.2.2 代码开发

    package.xml

    1. <assembly
    2.         xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
    3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.         xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
    5.     <formats>
    6.         <format>dir</format>
    7.     </formats>
    8.     <includeBaseDirectory>false</includeBaseDirectory>
    9.     <fileSets>
    10.         <fileSet>
    11.             <directory>src/main/resources</directory>
    12.             <includes>
    13.                 <include>plugin.json</include>
    14.                 <include>plugin_job_template.json</include>
    15.             </includes>
    16.             <outputDirectory>plugin/writer/s3writer</outputDirectory>
    17.         </fileSet>
    18.         <fileSet>
    19.             <directory>target/</directory>
    20.             <includes>
    21.                 <include>s3writer-0.0.1-SNAPSHOT.jar</include>
    22.             </includes>
    23.             <outputDirectory>plugin/writer/s3writer</outputDirectory>
    24.         </fileSet>
    25.     </fileSets>
    26.  
    27.     <dependencySets>
    28.         <dependencySet>
    29.             <useProjectArtifact>false</useProjectArtifact>
    30.             <outputDirectory>plugin/writer/s3writer/libs</outputDirectory>
    31.             <scope>runtime</scope>
    32.         </dependencySet>
    33.     </dependencySets>
    34. </assembly>

    Constant.java

    1. package com.alibaba.datax.plugin.writer.s3writer;
    2.  
    3.  
    4. public class Constant {
    5.     public static final String OBJECT = "object";
    6.     public static final int CONNECT_TIMEOUT = 50000;
    7.     public static final String SIGNER_OVERRIDE = "S3SignerType";
    8. }

    Key.java

    1. package com.alibaba.datax.plugin.writer.s3writer;
    2.  
    3. public class Key {
    4.     public static final String ENDPOINT = "endpoint";
    5.  
    6.     public static final String ACCESSKEY = "accessKey";
    7.  
    8.     public static final String SECRETKEY = "secretKey";
    9.  
    10.     public static final String BUCKET = "bucket";
    11.  
    12.     public static final String OBJECT = "object";
    13.  
    14. }

    S3Util.java 如上

    S3Writer.java

    1. package com.alibaba.datax.plugin.writer.s3writer;
    2.  
    3. import com.alibaba.datax.common.element.Record;
    4. import com.alibaba.datax.common.exception.DataXException;
    5. import com.alibaba.datax.common.plugin.RecordReceiver;
    6. import com.alibaba.datax.common.spi.Writer;
    7. import com.alibaba.datax.common.util.Configuration;
    8. import com.alibaba.datax.common.util.RetryUtil;
    9. import com.alibaba.datax.plugin.unstructuredstorage.writer.TextCsvWriterManager;
    10. import com.alibaba.datax.plugin.unstructuredstorage.writer.UnstructuredStorageWriterUtil;
    11. import com.alibaba.datax.plugin.unstructuredstorage.writer.UnstructuredWriter;
    12. import com.amazonaws.services.s3.AmazonS3;
    13. import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
    14. import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
    15. import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
    16. import com.amazonaws.services.s3.model.InitiateMultipartUploadResult;
    17. import com.amazonaws.services.s3.model.ObjectListing;
    18. import com.amazonaws.services.s3.model.PartETag;
    19. import com.amazonaws.services.s3.model.S3ObjectSummary;
    20. import com.amazonaws.services.s3.model.UploadPartRequest;
    21. import com.amazonaws.services.s3.model.UploadPartResult;
    22. import org.apache.commons.io.IOUtils;
    23. import org.apache.commons.lang3.StringUtils;
    24. import org.slf4j.Logger;
    25. import org.slf4j.LoggerFactory;
    26.  
    27. import java.io.ByteArrayInputStream;
    28. import java.io.IOException;
    29. import java.io.InputStream;
    30. import java.io.StringWriter;
    31. import java.text.DateFormat;
    32. import java.text.SimpleDateFormat;
    33. import java.util.ArrayList;
    34. import java.util.HashSet;
    35. import java.util.List;
    36. import java.util.Set;
    37. import java.util.UUID;
    38. import java.util.concurrent.Callable;
    39.  
    40.  
    41. public class S3Writer extends Writer {
    42.     public static class Job extends Writer.Job {
    43.         private static final Logger LOG = LoggerFactory.getLogger(Job.class);
    44.  
    45.         private Configuration writerSliceConfig = null;
    46.         private AmazonS3 s3client = null;
    47.  
    48.         @Override
    49.         public void init() {
    50.             this.writerSliceConfig = this.getPluginJobConf();
    51.             this.validateParameter();
    52.             this.s3client = S3Util.initS3Client(this.writerSliceConfig);
    53.         }
    54.  
    55.         private void validateParameter() {
    56.             this.writerSliceConfig.getNecessaryValue(Key.ENDPOINT,
    57.                     S3WriterErrorCode.REQUIRED_VALUE);
    58.             this.writerSliceConfig.getNecessaryValue(Key.SECRETKEY,
    59.                     S3WriterErrorCode.REQUIRED_VALUE);
    60.             this.writerSliceConfig.getNecessaryValue(Key.ACCESSKEY,
    61.                     S3WriterErrorCode.REQUIRED_VALUE);
    62.             this.writerSliceConfig.getNecessaryValue(Key.BUCKET,
    63.                     S3WriterErrorCode.REQUIRED_VALUE);
    64.             this.writerSliceConfig.getNecessaryValue(Key.OBJECT,
    65.                     S3WriterErrorCode.REQUIRED_VALUE);
    66.  
    67.             // warn: do not support compress!!
    68.             String compress = this.writerSliceConfig.getString(com.alibaba.datax.plugin.unstructuredstorage.writer.Key.COMPRESS);
    69.             if (StringUtils.isNotBlank(compress)) {
    70.                 String errorMessage = String.format("S3写暂时不支持压缩, 该压缩配置项[%s]不起效用", compress);
    71.                 LOG.error(errorMessage);
    72.                 throw DataXException.asDataXException(S3WriterErrorCode.ILLEGAL_VALUE, errorMessage);
    73.             }
    74.  
    75.             UnstructuredStorageWriterUtil.validateParameter(this.writerSliceConfig);
    76.         }
    77.  
    78.         @Override
    79.         public void prepare() {
    80.             LOG.info("begin do prepare...");
    81.             String bucket = this.writerSliceConfig.getString(Key.BUCKET);
    82.             String object = this.writerSliceConfig.getString(Key.OBJECT);
    83.             String writeMode = this.writerSliceConfig
    84.                     .getString(com.alibaba.datax.plugin.unstructuredstorage.writer.Key.WRITE_MODE);
    85.             // warn: bucket is not exists, create it
    86.             try {
    87.                 // warn: do not create bucket for user
    88.                 if (!this.s3client.doesBucketExist(bucket)) {
    89.                     // this.s3client.createBucket(bucket);
    90.                     String errorMessage = String.format(
    91.                             "您配置的bucket [%s] 不存在, 请您确认您的配置项.", bucket);
    92.                     LOG.error(errorMessage);
    93.                     throw DataXException.asDataXException(
    94.                             S3WriterErrorCode.ILLEGAL_VALUE, errorMessage);
    95.                 }
    96.                 LOG.info(String.format("access control details [%s].",
    97.                         this.s3client.getBucketAcl(bucket).toString()));
    98.  
    99.                 // truncate option handler
    100.                 if ("truncate".equals(writeMode)) {
    101.                     LOG.info(String
    102.                             .format("由于您配置了writeMode truncate, 开始清理 [%s] 下面以 [%s] 开头的Object",
    103.                                     bucket, object));
    104.                     // warn: 默认情况下，如果Bucket中的Object数量大于100，则只会返回100个Object
    105.                     while (true) {
    106.                         ObjectListing listing = null;
    107.                         LOG.info("list objects with listObject(bucket, object)");
    108.                         listing = this.s3client.listObjects(bucket, object);
    109.                         List<S3ObjectSummary> objectSummarys = listing.getObjectSummaries();
    110.                         for (S3ObjectSummary objectSummary : objectSummarys) {
    111.                             LOG.info(String.format("delete oss object [%s].", objectSummary.getKey()));
    112.                             this.s3client.deleteObject(bucket, objectSummary.getKey());
    113.                         }
    114.                         if (objectSummarys.isEmpty()) {
    115.                             break;
    116.                         }
    117.                     }
    118.                 } else if ("append".equals(writeMode)) {
    119.                     LOG.info(String
    120.                             .format("由于您配置了writeMode append, 写入前不做清理工作, 数据写入Bucket [%s] 下, 写入相应Object的前缀为  [%s]",
    121.                                     bucket, object));
    122.                 } else if ("nonConflict".equals(writeMode)) {
    123.                     LOG.info(String
    124.                             .format("由于您配置了writeMode nonConflict, 开始检查Bucket [%s] 下面以 [%s] 命名开头的Object",
    125.                                     bucket, object));
    126.                     ObjectListing listing = this.s3client.listObjects(bucket,
    127.                             object);
    128.                     if (0 < listing.getObjectSummaries().size()) {
    129.                         StringBuilder objectKeys = new StringBuilder();
    130.                         objectKeys.append("[ ");
    131.                         for (S3ObjectSummary ossObjectSummary : listing.getObjectSummaries()) {
    132.                             objectKeys.append(ossObjectSummary.getKey() + " ,");
    133.                         }
    134.                         objectKeys.append(" ]");
    135.                         LOG.info(String.format("object with prefix [%s] details: %s", object, objectKeys.toString()));
    136.                         throw DataXException.asDataXException(
    137.                                 S3WriterErrorCode.ILLEGAL_VALUE,
    138.                                 String.format("您配置的Bucket: [%s] 下面存在其Object有前缀 [%s].", bucket, object));
    139.                     }
    140.                 }
    141.             } catch (Exception e) {
    142.                 throw DataXException.asDataXException(
    143.                         S3WriterErrorCode.S3_COMM_ERROR, e.getMessage());
    144.             }
    145.         }
    146.  
    147.         @Override
    148.         public void post() {
    149.  
    150.         }
    151.  
    152.         @Override
    153.         public void destroy() {
    154.  
    155.         }
    156.  
    157.         @Override
    158.         public List<Configuration> split(int mandatoryNumber) {
    159.             LOG.info("begin do split...");
    160.             List<Configuration> writerSplitConfigs = new ArrayList<Configuration>();
    161.             String object = this.writerSliceConfig.getString(Key.OBJECT);
    162.             String bucket = this.writerSliceConfig.getString(Key.BUCKET);
    163.  
    164.             Set<String> allObjects = new HashSet<String>();
    165.             try {
    166.                 List<S3ObjectSummary> ossObjectlisting = this.s3client.listObjects(bucket).getObjectSummaries();
    167.                 for (S3ObjectSummary objectSummary : ossObjectlisting) {
    168.                     allObjects.add(objectSummary.getKey());
    169.                 }
    170.             } catch (Exception e) {
    171.                 throw DataXException.asDataXException(
    172.                         S3WriterErrorCode.S3_COMM_ERROR, e.getMessage());
    173.             }
    174.  
    175.             String objectSuffix;
    176.             for (int i = 0; i < mandatoryNumber; i++) {
    177.                 // handle same object name
    178.                 Configuration splitedTaskConfig = this.writerSliceConfig
    179.                         .clone();
    180.  
    181.                 String fullObjectName = null;
    182.                 objectSuffix = StringUtils.replace(
    183.                         UUID.randomUUID().toString(), "-", "");
    184.                 fullObjectName = String.format("%s__%s", object, objectSuffix);
    185.                 while (allObjects.contains(fullObjectName)) {
    186.                     objectSuffix = StringUtils.replace(UUID.randomUUID()
    187.                             .toString(), "-", "");
    188.                     fullObjectName = String.format("%s__%s", object,
    189.                             objectSuffix);
    190.                 }
    191.                 allObjects.add(fullObjectName);
    192.  
    193.                 splitedTaskConfig.set(Key.OBJECT, fullObjectName);
    194.  
    195.                 LOG.info(String.format("splited write object name:[%s]",
    196.                         fullObjectName));
    197.  
    198.                 writerSplitConfigs.add(splitedTaskConfig);
    199.             }
    200.             LOG.info("end do split.");
    201.             return writerSplitConfigs;
    202.         }
    203.     }
    204.  
    205.     public static class Task extends Writer.Task {
    206.         private static final Logger LOG = LoggerFactory.getLogger(Task.class);
    207.  
    208.         private AmazonS3 s3client;
    209.         private Configuration writerSliceConfig;
    210.         private String bucket;
    211.         private String object;
    212.         private String nullFormat;
    213.         private String encoding;
    214.         private char fieldDelimiter;
    215.         private String dateFormat;
    216.         private DateFormat dateParse;
    217.         private String fileFormat;
    218.         private List<String> header;
    219.         private Long maxFileSize;// MB
    220.         private String suffix;
    221.  
    222.         @Override
    223.         public void init() {
    224.             this.writerSliceConfig = this.getPluginJobConf();
    225.             this.s3client = S3Util.initS3Client(this.writerSliceConfig);
    226.             this.bucket = this.writerSliceConfig.getString(Key.BUCKET);
    227.             this.object = this.writerSliceConfig.getString(Key.OBJECT);
    228.             this.nullFormat = this.writerSliceConfig
    229.                     .getString(com.alibaba.datax.plugin.unstructuredstorage.writer.Key.NULL_FORMAT);
    230.             this.dateFormat = this.writerSliceConfig
    231.                     .getString(
    232.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Key.DATE_FORMAT,
    233.                             null);
    234.             if (StringUtils.isNotBlank(this.dateFormat)) {
    235.                 this.dateParse = new SimpleDateFormat(dateFormat);
    236.             }
    237.             this.encoding = this.writerSliceConfig
    238.                     .getString(
    239.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Key.ENCODING,
    240.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Constant.DEFAULT_ENCODING);
    241.             this.fieldDelimiter = this.writerSliceConfig
    242.                     .getChar(
    243.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Key.FIELD_DELIMITER,
    244.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Constant.DEFAULT_FIELD_DELIMITER);
    245.             this.fileFormat = this.writerSliceConfig
    246.                     .getString(
    247.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Key.FILE_FORMAT,
    248.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Constant.FILE_FORMAT_TEXT);
    249.             this.header = this.writerSliceConfig
    250.                     .getList(
    251.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Key.HEADER,
    252.                             null, String.class);
    253.             this.maxFileSize = this.writerSliceConfig
    254.                     .getLong(
    255.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Key.MAX_FILE_SIZE,
    256.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Constant.MAX_FILE_SIZE);
    257.             this.suffix = this.writerSliceConfig
    258.                     .getString(
    259.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Key.SUFFIX,
    260.                             com.alibaba.datax.plugin.unstructuredstorage.writer.Constant.DEFAULT_SUFFIX);
    261.             this.suffix = this.suffix.trim();// warn: need trim
    262.         }
    263.  
    264.         @Override
    265.         public void startWrite(RecordReceiver lineReceiver) {
    266.             // 设置每块字符串长度
    267.             final long partSize = 1024 * 1024 * 10L;
    268.             long numberCacul = (this.maxFileSize * 1024 * 1024L) / partSize;
    269.             final long maxPartNumber = numberCacul >= 1 ? numberCacul : 1;
    270.             int objectRollingNumber = 0;
    271.             //warn: may be StringBuffer->StringBuilder
    272.             StringWriter sw = new StringWriter();
    273.             StringBuffer sb = sw.getBuffer();
    274.             UnstructuredWriter unstructuredWriter = TextCsvWriterManager
    275.                     .produceUnstructuredWriter(this.fileFormat,
    276.                             this.fieldDelimiter, sw);
    277.             Record record = null;
    278.  
    279.             LOG.info(String.format(
    280.                     "begin do write, each object maxFileSize: [%s]MB...",
    281.                     maxPartNumber * 10));
    282.             String currentObject = this.object;
    283.             InitiateMultipartUploadRequest currentInitiateMultipartUploadRequest = null;
    284.             InitiateMultipartUploadResult currentInitiateMultipartUploadResult = null;
    285.             boolean gotData = false;
    286.             List<PartETag> currentPartETags = null;
    287.             // to do:
    288.             // 可以根据currentPartNumber做分块级别的重试，InitiateMultipartUploadRequest多次一个currentPartNumber会覆盖原有
    289.             int currentPartNumber = 1;
    290.             try {
    291.                 // warn
    292.                 boolean needInitMultipartTransform = true;
    293.                 while ((record = lineReceiver.getFromReader()) != null) {
    294.                     gotData = true;
    295.                     // init:begin new multipart upload
    296.                     if (needInitMultipartTransform) {
    297.                         if (objectRollingNumber == 0) {
    298.                             if (StringUtils.isBlank(this.suffix)) {
    299.                                 currentObject = this.object;
    300.                             } else {
    301.                                 currentObject = String.format("%s%s",
    302.                                         this.object, this.suffix);
    303.                             }
    304.                         } else {
    305.                             // currentObject is like(no suffix)
    306.                             // myfile__9b886b70fbef11e59a3600163e00068c_1
    307.                             if (StringUtils.isBlank(this.suffix)) {
    308.                                 currentObject = String.format("%s_%s",
    309.                                         this.object, objectRollingNumber);
    310.                             } else {
    311.                                 // or with suffix
    312.                                 // myfile__9b886b70fbef11e59a3600163e00068c_1.csv
    313.                                 currentObject = String.format("%s_%s%s",
    314.                                         this.object, objectRollingNumber,
    315.                                         this.suffix);
    316.                             }
    317.                         }
    318.                         objectRollingNumber++;
    319.                         currentInitiateMultipartUploadRequest = new InitiateMultipartUploadRequest(
    320.                                 this.bucket, currentObject);
    321.                         currentInitiateMultipartUploadResult = this.s3client
    322.                                 .initiateMultipartUpload(currentInitiateMultipartUploadRequest);
    323.                         currentPartETags = new ArrayList<PartETag>();
    324.                         LOG.info(String
    325.                                 .format("write to bucket: [%s] object: [%s] with oss uploadId: [%s]",
    326.                                         this.bucket, currentObject,
    327.                                         currentInitiateMultipartUploadResult
    328.                                                 .getUploadId()));
    329.  
    330.                         // each object's header
    331.                         if (null != this.header && !this.header.isEmpty()) {
    332.                             unstructuredWriter.writeOneRecord(this.header);
    333.                         }
    334.                         // warn
    335.                         needInitMultipartTransform = false;
    336.                         currentPartNumber = 1;
    337.                     }
    339.                     // write: upload data to current object
    340.                     UnstructuredStorageWriterUtil.transportOneRecord(record,
    341.                             this.nullFormat, this.dateParse,
    342.                             this.getTaskPluginCollector(), unstructuredWriter);
    344.                     if (sb.length() >= partSize) {
    345.                         this.uploadOnePart(sw, currentPartNumber,
    346.                                 currentInitiateMultipartUploadResult,
    347.                                 currentPartETags, currentObject);
    348.                         currentPartNumber++;
    349.                         sb.setLength(0);
    350.                     }
    352.                     // save: end current multipart upload
    353.                     if (currentPartNumber > maxPartNumber) {
    354.                         LOG.info(String
    355.                                 .format("current object [%s] size > %s, complete current multipart upload and begin new one",
    356.                                         currentObject, currentPartNumber
    357.                                                 * partSize));
    358.                         CompleteMultipartUploadRequest currentCompleteMultipartUploadRequest = new CompleteMultipartUploadRequest(
    359.                                 this.bucket, currentObject,
    360.                                 currentInitiateMultipartUploadResult
    361.                                         .getUploadId(), currentPartETags);
    362.                         CompleteMultipartUploadResult currentCompleteMultipartUploadResult = this.s3client
    363.                                 .completeMultipartUpload(currentCompleteMultipartUploadRequest);
    364.                         LOG.info(String.format(
    365.                                 "final object [%s] etag is:[%s]",
    366.                                 currentObject,
    367.                                 currentCompleteMultipartUploadResult.getETag()));
    368.                         // warn
    369.                         needInitMultipartTransform = true;
    370.                     }
    371.                 }
    373.                 if (!gotData) {
    374.                     LOG.info("Receive no data from the source.");
    375.                     currentInitiateMultipartUploadRequest = new InitiateMultipartUploadRequest(
    376.                             this.bucket, currentObject);
    377.                     currentInitiateMultipartUploadResult = this.s3client
    378.                             .initiateMultipartUpload(currentInitiateMultipartUploadRequest);
    379.                     currentPartETags = new ArrayList();
    380.                     // each object's header
    381.                     if (null != this.header && !this.header.isEmpty()) {
    382.                         unstructuredWriter.writeOneRecord(this.header);
    383.                     }
    384.                 }
    385.                 // warn: may be some data stall in sb
    386.                 if (0 < sb.length()) {
    387.                     this.uploadOnePart(sw, currentPartNumber,
    388.                             currentInitiateMultipartUploadResult,
    389.                             currentPartETags, currentObject);
    390.                 }
    391.                 CompleteMultipartUploadRequest completeMultipartUploadRequest = new CompleteMultipartUploadRequest(
    392.                         this.bucket, currentObject,
    393.                         currentInitiateMultipartUploadResult.getUploadId(),
    394.                         currentPartETags);
    395.                 CompleteMultipartUploadResult completeMultipartUploadResult = this.s3client
    396.                         .completeMultipartUpload(completeMultipartUploadRequest);
    397.                 LOG.info(String.format("final object etag is:[%s]",
    398.                         completeMultipartUploadResult.getETag()));
    399.             } catch (IOException e) {
    400.                 // 脏数据UnstructuredStorageWriterUtil.transportOneRecord已经记录,header
    401.                 // 都是字符串不认为有脏数据
    402.                 throw DataXException.asDataXException(
    403.                         S3WriterErrorCode.Write_OBJECT_ERROR, e.getMessage());
    404.             } catch (Exception e) {
    405.                 throw DataXException.asDataXException(
    406.                         S3WriterErrorCode.Write_OBJECT_ERROR, e.getMessage());
    407.             }
    408.             LOG.info("end do write");
    409.         }
    410.  
    411.         /**
    412.          * 对于同一个UploadID，该号码不但唯一标识这一块数据，也标识了这块数据在整个文件内的相对位置。
    413.          * 如果你用同一个part号码，上传了新的数据，那么OSS上已有的这个号码的Part数据将被覆盖。
    414.          *
    415.          * @throws Exception
    416.          */
    417.         private void uploadOnePart(
    418.                 final StringWriter sw,
    419.                 final int partNumber,
    420.                 final InitiateMultipartUploadResult initiateMultipartUploadResult,
    421.                 final List<PartETag> partETags, final String currentObject)
    422.                 throws Exception {
    423.             final String encoding = this.encoding;
    424.             final String bucket = this.bucket;
    425.             final AmazonS3 s3client = this.s3client;
    426.             RetryUtil.executeWithRetry(new Callable<Boolean>() {
    427.                 @Override
    428.                 public Boolean call() throws Exception {
    429.                     byte[] byteArray = sw.toString().getBytes(encoding);
    430.                     InputStream inputStream = new ByteArrayInputStream(
    431.                             byteArray);
    432.                     // 创建UploadPartRequest，上传分块
    433.                     UploadPartRequest uploadPartRequest = new UploadPartRequest();
    434.                     uploadPartRequest.setBucketName(bucket);
    435.                     uploadPartRequest.setKey(currentObject);
    436.                     uploadPartRequest.setUploadId(initiateMultipartUploadResult
    437.                             .getUploadId());
    438.                     uploadPartRequest.setInputStream(inputStream);
    439.                     uploadPartRequest.setPartSize(byteArray.length);
    440.                     uploadPartRequest.setPartNumber(partNumber);
    441.                     UploadPartResult uploadPartResult = s3client
    442.                             .uploadPart(uploadPartRequest);
    443.                     partETags.add(uploadPartResult.getPartETag());
    444.                     LOG.info(String
    445.                             .format("upload part [%s] size [%s] Byte has been completed.",
    446.                                     partNumber, byteArray.length));
    447.                     IOUtils.closeQuietly(inputStream);
    448.                     return true;
    449.                 }
    450.             }, 3, 1000L, false);
    451.         }
    452.  
    453.         @Override
    454.         public void prepare() {
    455.  
    456.         }
    457.  
    458.         @Override
    459.         public void post() {
    460.  
    461.         }
    462.  
    463.         @Override
    464.         public void destroy() {
    465.  
    466.         }
    467.     }
    468. }

    S3WriterErrorCode.java

    1. package com.alibaba.datax.plugin.writer.s3writer;
    2.  
    3. import com.alibaba.datax.common.spi.ErrorCode;
    4.  
    5. public enum S3WriterErrorCode implements ErrorCode {
    6.  
    7.     CONFIG_INVALID_EXCEPTION("S3Writer-00", "您的参数配置错误."),
    8.     REQUIRED_VALUE("S3Writer-01", "您缺失了必须填写的参数值."),
    9.     ILLEGAL_VALUE("S3Writer-02", "您填写的参数值不合法."),
    10.     Write_OBJECT_ERROR("S3Writer-03", "您配置的目标Object在写入时异常."),
    11.     S3_COMM_ERROR("S3Writer-05", "执行相应的S3操作异常."),
    12.     ;
    13.  
    14.     private final String code;
    15.     private final String description;
    16.  
    17.     private S3WriterErrorCode(String code, String description) {
    18.         this.code = code;
    19.         this.description = description;
    20.     }
    21.  
    22.     @Override
    23.     public String getCode() {
    24.         return this.code;
    25.     }
    26.  
    27.     @Override
    28.     public String getDescription() {
    29.         return this.description;
    30.     }
    31.  
    32.     @Override
    33.     public String toString() {
    34.         return String.format("Code:[%s], Description:[%s].", this.code,
    35.                 this.description);
    36.     }
    37.  
    38. }

    plugin.json

    1. {
    2.     "name": "s3writer",
    3.     "class": "com.alibaba.datax.plugin.writer.s3writer.S3Writer",
    4.     "description": "",
    5.     "developer": "chenweifeng"
    6. }
    7.

    3 调试运行和性能测试

    s3reader 和 s3writer 模板

    略

    s3reader 和 s3writer 运行结果

  • 相关阅读:
    bean的生命周期
    操作系统_多线程笔记(二)
    【问题解决】新版webots纹理等资源文件加载/下载时间过长
    【LeetCode】No.55. Jump Game -- Java Version
    成功项目经理总结的20个项目管理经验
    PMP模拟题 | 每日一练，快速提分
    GPT-4o：人工智能新纪元的突破与展望
    Ubuntu安装和配置ssh
    Java 包及访问控制权限
    c语言进阶部分详解（指针进阶1）
  • 原文地址：https://blog.csdn.net/Carson073/article/details/128001868