建议新增数据同步到paimon支持S3、OSS

[kevinlin299]

目前tis从其他数据库批量或者实时同步到paimon只有hive可选，希望可以实现S3、OSS等，希望tis越来越好

使用的是自建minio 文件系统，

使用https://paimon.apache.org/docs/master/concepts/catalog/ 中的 filesystem catalog类型

参照文档：https://mp.weixin.qq.com/s/6sX2UArgjl6-3G6V7EWMdg?mpshare=1&scene=1&srcid=01126hpVQYzjiHl1KpitLYoW&sharer_shareinfo=65b5dd2648c83311130166dbc78bef36&sharer_shareinfo_first=65b5dd2648c83311130166dbc78bef36&version=5.0.3.70620&platform=mac#rd

[baisui1981]

目前tis从其他数据库批量或者实时同步到paimon只有hive可选，希望可以实现S3、OSS等，希望tis越来越好

你们现在是在亚马逊云上哪个区进行s3的测试？

亚马逊s3相关：

• 介绍： https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/userguide/UsingObjects.html
• 亚马逊云中国：https://www.amazonaws.cn/

@kevinlin299 方便的话留个微信号，交流一下

[baisui1981]

亚马逊AWS提供了完整的Java SDK来操作S3服务。以下是使用AWS Java SDK操作S3的详细指南：

1. 添加依赖

Maven依赖：

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>s3</artifactId>
    <version>2.20.0</version> <!-- 使用最新版本 -->
</dependency>

Gradle依赖：

implementation 'software.amazon.awssdk:s3:2.20.0'

2. 基础配置

import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;

public class S3Example {
    private static S3Client s3Client;
    
    static {
        // 方式1：使用IAM角色（推荐生产环境）
        s3Client = S3Client.builder()
                .region(Region.AP_NORTHEAST_1) // 根据实际情况修改
                .build();
                
        // 方式2：使用Access Key（开发环境）
        AwsBasicCredentials awsCreds = AwsBasicCredentials.create(
            "your-access-key-id",
            "your-secret-access-key"
        );
        
        s3Client = S3Client.builder()
                .region(Region.AP_NORTHEAST_1)
                .credentialsProvider(StaticCredentialsProvider.create(awsCreds))
                .build();
    }
}

3. 常用操作示例

3.1 列出存储桶

public void listBuckets() {
    ListBucketsResponse response = s3Client.listBuckets();
    response.buckets().forEach(bucket -> {
        System.out.println("Bucket: " + bucket.name());
    });
}

3.2 创建存储桶

public void createBucket(String bucketName) {
    CreateBucketRequest request = CreateBucketRequest.builder()
            .bucket(bucketName)
            .build();
    s3Client.createBucket(request);
    System.out.println("Bucket created: " + bucketName);
}

3.3 上传文件

public void uploadFile(String bucketName, String key, File file) {
    PutObjectRequest request = PutObjectRequest.builder()
            .bucket(bucketName)
            .key(key)
            .contentType("application/octet-stream")
            .build();
    
    s3Client.putObject(request, file.toPath());
    System.out.println("File uploaded: " + key);
}

// 使用流上传
public void uploadInputStream(String bucketName, String key, 
                             InputStream input, long contentLength) {
    PutObjectRequest request = PutObjectRequest.builder()
            .bucket(bucketName)
            .key(key)
            .build();
            
    s3Client.putObject(request,
        RequestBody.fromInputStream(input, contentLength));
}

3.4 下载文件

public void downloadFile(String bucketName, String key, String destinationPath) {
    GetObjectRequest request = GetObjectRequest.builder()
            .bucket(bucketName)
            .key(key)
            .build();
    
    ResponseInputStream<GetObjectResponse> response = 
        s3Client.getObject(request);
    
    Files.copy(response, Paths.get(destinationPath));
    response.close();
}

3.5 列出对象

public void listObjects(String bucketName) {
    ListObjectsV2Request request = ListObjectsV2Request.builder()
            .bucket(bucketName)
            .maxKeys(10)
            .build();
    
    ListObjectsV2Response response;
    do {
        response = s3Client.listObjectsV2(request);
        response.contents().forEach(obj -> {
            System.out.println("Object: " + obj.key() + 
                             " Size: " + obj.size());
        });
        
        // 处理分页
        request = request.toBuilder()
                .continuationToken(response.nextContinuationToken())
                .build();
    } while (response.isTruncated());
}

3.6 删除对象

public void deleteObject(String bucketName, String key) {
    DeleteObjectRequest request = DeleteObjectRequest.builder()
            .bucket(bucketName)
            .key(key)
            .build();
    
    s3Client.deleteObject(request);
    System.out.println("Object deleted: " + key);
}

3.7 生成预签名URL

public String generatePresignedUrl(String bucketName, String key, 
                                  Duration expiration) {
    GetObjectRequest getObjectRequest = GetObjectRequest.builder()
            .bucket(bucketName)
            .key(key)
            .build();
    
    return s3Client.utilities()
            .getUrl(builder -> builder
                .getObjectRequest(getObjectRequest)
                .expiration(expiration))
            .toString();
}

4. 高级功能

4.1 分片上传（大文件）

public void multipartUpload(String bucketName, String key, 
                           Path filePath) throws Exception {
    
    CreateMultipartUploadRequest createRequest = 
        CreateMultipartUploadRequest.builder()
            .bucket(bucketName)
            .key(key)
            .build();
    
    CreateMultipartUploadResponse response = 
        s3Client.createMultipartUpload(createRequest);
    String uploadId = response.uploadId();
    
    // 上传分片逻辑
    // ...
    
    // 完成上传
    CompleteMultipartUploadRequest completeRequest = 
        CompleteMultipartUploadRequest.builder()
            .bucket(bucketName)
            .key(key)
            .uploadId(uploadId)
            .build();
    
    s3Client.completeMultipartUpload(completeRequest);
}

4.2 使用S3 TransferManager（简化操作）

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>s3-transfer-manager</artifactId>
    <version>2.20.0</version>
</dependency>

import software.amazon.awssdk.transfer.s3.S3TransferManager;

S3TransferManager transferManager = S3TransferManager.builder()
        .s3Client(s3Client)
        .build();

// 上传大文件
FileUpload upload = transferManager.uploadFile(builder -> builder
        .source(Paths.get("largefile.zip"))
        .putObjectRequest(req -> req
            .bucket(bucketName)
            .key("largefile.zip"))
        .build());

upload.completionFuture().join();

5. 最佳实践

1. 客户端管理：重用S3Client实例，避免频繁创建
2. 错误处理：

try {
    s3Client.getObject(request);
} catch (NoSuchKeyException e) {
    // 对象不存在
} catch (S3Exception e) {
    // 其他S3错误
    System.err.println(e.awsErrorDetails().errorMessage());
}

3. 连接配置：

S3Client.builder()
    .region(Region.AP_NORTHEAST_1)
    .httpClientBuilder(ApacheHttpClient.builder()
        .maxConnections(100)
        .connectionTimeout(Duration.ofSeconds(10)))
    .build();

6. 官方文档和资源

• AWS SDK for Java 2.x官方文档
• S3 Java SDK API参考
• GitHub示例代码

注意事项：

1. 确保AWS凭证安全，不要硬编码在代码中
2. 生产环境建议使用IAM角色进行权限管理
3. 注意SDK版本，v1和v2的API有较大差异
4. 处理大文件时考虑使用分片上传
5. 合理设置重试策略和超时时间

这是基本的S3 Java SDK操作指南，你可以根据具体需求选择合适的方法。

[baisui1981]

以上这个sdk 也支持 minio的文件系统吗？

是的，AWS S3 Java SDK 也支持 MinIO，因为 MinIO 完全兼容 S3 API。 以下是详细的配置和使用方法：

1. 基础配置连接 MinIO

import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.S3Configuration;
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;

import java.net.URI;
import java.time.Duration;

public class MinioS3Client {
    
    public static S3Client createMinioClient() {
        // MinIO 服务器地址
        URI minioEndpoint = URI.create("http://localhost:9000");
        
        // MinIO 的 Access Key 和 Secret Key
        // 可以在启动 MinIO 时设置，或通过 MINIO_ROOT_USER/MINIO_ROOT_PASSWORD 环境变量
        AwsBasicCredentials credentials = AwsBasicCredentials.create(
            "your-minio-access-key",  // 通常是 MINIO_ROOT_USER
            "your-minio-secret-key"   // 通常是 MINIO_ROOT_PASSWORD
        );

        // 创建 S3 客户端配置
        return S3Client.builder()
                .endpointOverride(minioEndpoint)  // 关键：覆盖为 MinIO 端点
                .credentialsProvider(StaticCredentialsProvider.create(credentials))
                .region(Region.US_EAST_1)  // MinIO 不需要特定区域，但必须设置一个
                .serviceConfiguration(S3Configuration.builder()
                        .pathStyleAccessEnabled(true)  // 关键：启用路径风格访问
                        .build())
                .overrideConfiguration(ClientOverrideConfiguration.builder()
                        // 增加超时时间
                        .apiCallTimeout(Duration.ofMinutes(2))
                        .apiCallAttemptTimeout(Duration.ofMinutes(1))
                        .build())
                .build();
    }
}

2. 完整使用示例

import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.model.*;
import java.io.File;
import java.util.List;

public class MinioS3Example {
    private static final S3Client s3Client;
    
    static {
        s3Client = createMinioClient();
    }
    
    private static S3Client createMinioClient() {
        // 从环境变量或配置文件中读取 MinIO 配置
        String endpoint = System.getenv("MINIO_ENDPOINT") != null ? 
                System.getenv("MINIO_ENDPOINT") : "http://localhost:9000";
        String accessKey = System.getenv("MINIO_ACCESS_KEY") != null ? 
                System.getenv("MINIO_ACCESS_KEY") : "minioadmin";
        String secretKey = System.getenv("MINIO_SECRET_KEY") != null ? 
                System.getenv("MINIO_SECRET_KEY") : "minioadmin";
        
        return S3Client.builder()
                .endpointOverride(URI.create(endpoint))
                .credentialsProvider(StaticCredentialsProvider.create(
                        AwsBasicCredentials.create(accessKey, secretKey)))
                .region(Region.US_EAST_1)
                .serviceConfiguration(S3Configuration.builder()
                        .pathStyleAccessEnabled(true)
                        .build())
                .build();
    }
    
    // 创建存储桶
    public void createBucket(String bucketName) {
        try {
            CreateBucketRequest request = CreateBucketRequest.builder()
                    .bucket(bucketName)
                    .build();
            
            s3Client.createBucket(request);
            System.out.println("Bucket created: " + bucketName);
        } catch (S3Exception e) {
            System.err.println("Error creating bucket: " + e.getMessage());
        }
    }
    
    // 上传文件到 MinIO
    public void uploadFile(String bucketName, String objectKey, File file) {
        try {
            PutObjectRequest request = PutObjectRequest.builder()
                    .bucket(bucketName)
                    .key(objectKey)
                    .build();
            
            s3Client.putObject(request, RequestBody.fromFile(file));
            System.out.println("File uploaded to MinIO: " + objectKey);
        } catch (S3Exception e) {
            System.err.println("Error uploading file: " + e.getMessage());
        }
    }
    
    // 列出存储桶中的所有对象
    public void listObjects(String bucketName) {
        try {
            ListObjectsV2Request request = ListObjectsV2Request.builder()
                    .bucket(bucketName)
                    .build();
            
            ListObjectsV2Response response = s3Client.listObjectsV2(request);
            response.contents().forEach(object -> {
                System.out.println("Object: " + object.key() + 
                                 " Size: " + object.size() + 
                                 " Last Modified: " + object.lastModified());
            });
        } catch (S3Exception e) {
            System.err.println("Error listing objects: " + e.getMessage());
        }
    }
    
    // 下载文件
    public void downloadFile(String bucketName, String objectKey, String destinationPath) {
        try {
            GetObjectRequest request = GetObjectRequest.builder()
                    .bucket(bucketName)
                    .key(objectKey)
                    .build();
            
            s3Client.getObject(request, java.nio.file.Paths.get(destinationPath));
            System.out.println("File downloaded from MinIO: " + destinationPath);
        } catch (S3Exception e) {
            System.err.println("Error downloading file: " + e.getMessage());
        }
    }
    
    // 删除对象
    public void deleteObject(String bucketName, String objectKey) {
        try {
            DeleteObjectRequest request = DeleteObjectRequest.builder()
                    .bucket(bucketName)
                    .key(objectKey)
                    .build();
            
            s3Client.deleteObject(request);
            System.out.println("Object deleted from MinIO: " + objectKey);
        } catch (S3Exception e) {
            System.err.println("Error deleting object: " + e.getMessage());
        }
    }
    
    // 检查存储桶是否存在
    public boolean bucketExists(String bucketName) {
        try {
            HeadBucketRequest request = HeadBucketRequest.builder()
                    .bucket(bucketName)
                    .build();
            
            s3Client.headBucket(request);
            return true;
        } catch (NoSuchBucketException e) {
            return false;
        } catch (S3Exception e) {
            System.err.println("Error checking bucket: " + e.getMessage());
            return false;
        }
    }
    
    public static void main(String[] args) {
        MinioS3Example example = new MinioS3Example();
        
        String bucketName = "my-bucket";
        String objectKey = "test-file.txt";
        File file = new File("local-file.txt");
        
        // 使用示例
        if (!example.bucketExists(bucketName)) {
            example.createBucket(bucketName);
        }
        
        example.uploadFile(bucketName, objectKey, file);
        example.listObjects(bucketName);
        example.downloadFile(bucketName, objectKey, "downloaded.txt");
    }
}

3. Spring Boot 集成配置

# application.yml
minio:
  endpoint: http://localhost:9000
  access-key: minioadmin
  secret-key: minioadmin
  region: us-east-1
  bucket: my-bucket

@Configuration
public class MinioConfig {
    
    @Value("${minio.endpoint}")
    private String endpoint;
    
    @Value("${minio.access-key}")
    private String accessKey;
    
    @Value("${minio.secret-key}")
    private String secretKey;
    
    @Value("${minio.region}")
    private String region;
    
    @Bean
    public S3Client s3Client() {
        return S3Client.builder()
                .endpointOverride(URI.create(endpoint))
                .credentialsProvider(StaticCredentialsProvider.create(
                        AwsBasicCredentials.create(accessKey, secretKey)))
                .region(Region.of(region))
                .serviceConfiguration(S3Configuration.builder()
                        .pathStyleAccessEnabled(true)
                        .build())
                .build();
    }
}

4. Gradle/Maven 依赖

Maven:

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>s3</artifactId>
    <version>2.20.0</version>
</dependency>

<!-- 如果需要异步支持 -->
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>s3-transfer-manager</artifactId>
    <version>2.20.0</version>
</dependency>

Gradle:

implementation 'software.amazon.awssdk:s3:2.20.0'
implementation 'software.amazon.awssdk:s3-transfer-manager:2.20.0'

5. 关键注意事项

5.1 必须启用的配置

.serviceConfiguration(S3Configuration.builder()
        .pathStyleAccessEnabled(true)  // 必须启用路径风格
        .build())

5.2 端点配置

• MinIO 本地开发: http://localhost:9000
• MinIO Docker: http://minio:9000 (容器内访问)
• MinIO HTTPS: https://minio.example.com

5.3 认证信息

• 默认 MinIO 安装的凭证通常是 minioadmin/minioadmin
• 生产环境一定要修改默认凭证

5.4 区域设置

MinIO 不需要特定的 AWS 区域，但 SDK 要求必须设置：

.region(Region.US_EAST_1)  // 任意区域都可以

5.5 兼容性检查

// 测试连接和兼容性
public boolean testMinioConnection() {
    try {
        s3Client.listBuckets();
        System.out.println("Successfully connected to MinIO");
        return true;
    } catch (Exception e) {
        System.err.println("Failed to connect to MinIO: " + e.getMessage());
        return false;
    }
}

6. 与原生 AWS S3 的区别

特性	AWS S3	MinIO + S3 SDK
端点	自动生成	必须显式设置 endpointOverride
路径风格	默认虚拟主机	必须启用 pathStyleAccessEnabled(true)
区域	必须正确设置	可任意设置
凭证	IAM/Access Key	MinIO 用户名/密码
功能支持	完整 S3 功能	大部分 S3 功能（部分高级功能可能不支持）

7. 故障排除

连接问题：

1. 检查 MinIO 服务状态：docker ps 或 systemctl status minio
2. 验证网络连接：curl http://localhost:9000/minio/health/live
3. 检查防火墙：确保 9000 端口开放

认证问题：

1. 验证凭证：使用 MinIO Web 控制台测试
2. 环境变量：确保正确设置了 MINIO_ROOT_USER 和 MINIO_ROOT_PASSWORD

SDK 配置问题：

1. 路径风格：确认 pathStyleAccessEnabled(true) 已设置
2. HTTPS：如果是 HTTPS，确保证书正确配置

8. 替代方案：MinIO 官方 Java SDK

如果你遇到兼容性问题，可以考虑使用 MinIO 官方 SDK：

<dependency>
    <groupId>io.minio</groupId>
    <artifactId>minio</artifactId>
    <version>8.5.2</version>
</dependency>

import io.minio.MinioClient;

MinioClient minioClient = MinioClient.builder()
        .endpoint("http://localhost:9000")
        .credentials("minioadmin", "minioadmin")
        .build();

总结

AWS S3 Java SDK 完全可以用于 MinIO，只需正确配置端点和路径风格。这种方式的好处是：

1. 代码复用：同一套代码可以用于 AWS S3 和 MinIO
2. API 一致性：使用标准 S3 API
3. 熟悉度：如果你熟悉 AWS SDK，学习成本低

主要记住两个关键配置：

1. endpointOverride() - 指向你的 MinIO 服务器
2. pathStyleAccessEnabled(true) - 启用路径风格访问

这样你就可以用 AWS S3 SDK 无缝操作 MinIO 存储了。