python import csv import json import os import shutil  def write(config, configured_catalog, input_messages):     try:         path = "/local/" + config["output_directory"]         if os.path.isdir(path):             shutil.rmtree(path)         os.makedirs(path)                  output_file = path + "/output.csv"                  with open(output_file, 'w+', newline='') as csvfile:             csvwriter = csv.writer(csvfile)             csvwriter.writerow(["stream", "data", "emitted_at"])                          for msg in input_messages:                 if str(msg.type) == "Type.RECORD":                     stream = msg.record.stream                     data = json.dumps(msg.record.data)                     emitted_at = msg.record.emitted_at                     csvwriter.writerow([stream, data, emitted_at])                     <http://logger.info|logger.info>(f"row written: {stream}, {data}, {emitted_at}")                  <http://logger.info|logger.info>(f"CSV file created at: {output_file}")     except Exception as e:         logger.error(f"Error writing to CSV: {str(e)}")         raise

python import boto3 import pandas as pd import io  def merge_csv_files_in_s3(bucket_name, prefix, output_key):     s3 = boto3.client('s3')          # Lấy danh sách các file CSV trong bucket     response = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)     all_files = [obj['Key'] for obj in response.get('Contents', []) if obj['Key'].endswith('.csv')]          # Đọc và gộp tất cả các file CSV     dfs = []     for file in all_files:         obj = s3.get_object(Bucket=bucket_name, Key=file)         df = pd.read_csv(io.BytesIO(obj['Body'].read()))         dfs.append(df)          # Gộp tất cả DataFrame     merged_df = pd.concat(dfs, ignore_index=True)          # Ghi DataFrame đã gộp vào một file CSV mới trong S3     csv_buffer = io.StringIO()     merged_df.to_csv(csv_buffer, index=False)     s3.put_object(Bucket=bucket_name, Key=output_key, Body=csv_buffer.getvalue())  # Sử dụng hàm merge_csv_files_in_s3('your-bucket-name', 'your-prefix/', 'output/merged_file.csv')

boto3

pandas

<s3://airbyte-dev-data/source_dp_indicator_daily_test_20220517/2022_06_18_13_b8dc2c44-6686-4386-87ff-84b82b111ce4/0.csv.gz> <s3://airbyte-dev-data/source_dp_indicator_daily_test_20220517/2022_06_18_13_b8dc2c44-6686-4386-87ff-84b82b111ce4/1.csv.gz> <s3://airbyte-dev-data/source_dp_indicator_daily_test_20220517/2022_06_18_13_b8dc2c44-6686-4386-87ff-84b82b111ce4/2.csv.gz> ...

write

python
import boto3
import csv
import io
from botocore.exceptions import ClientError

def write(config, configured_catalog, input_messages):
    s3_client = boto3.client(
        's3',
        aws_access_key_id=config['aws_access_key_id'],
        aws_secret_access_key=config['aws_secret_access_key']
    )
    bucket_name = config['bucket_name']
    object_key = config['object_key']

    # Khởi tạo multipart upload
    try:
        mpu = s3_client.create_multipart_upload(Bucket=bucket_name, Key=object_key)
        parts = []
        part_number = 1
        buffer = io.StringIO()
        csv_writer = csv.writer(buffer)

        for msg in input_messages:
            if msg.type == Type.RECORD:
                csv_writer.writerow([msg.record.stream, msg.record.data, msg.record.emitted_at])
                
                # Nếu buffer đạt kích thước nhất định, upload một phần
                if buffer.tell() > 5 * 1024 * 1024:  # 5MB
                    buffer.seek(0)
                    part = s3_client.upload_part(
                        Bucket=bucket_name,
                        Key=object_key,
                        PartNumber=part_number,
                        UploadId=mpu['UploadId'],
                        Body=buffer.getvalue().encode()
                    )
                    parts.append({"PartNumber": part_number, "ETag": part['ETag']})
                    part_number += 1
                    buffer = io.StringIO()
                    csv_writer = csv.writer(buffer)

        # Upload phần cuối cùng nếu còn dữ liệu trong buffer
        if buffer.tell() > 0:
            buffer.seek(0)
            part = s3_client.upload_part(
                Bucket=bucket_name,
                Key=object_key,
                PartNumber=part_number,
                UploadId=mpu['UploadId'],
                Body=buffer.getvalue().encode()
            )
            parts.append({"PartNumber": part_number, "ETag": part['ETag']})

        # Hoàn thành multipart upload
        s3_client.complete_multipart_upload(
            Bucket=bucket_name,
            Key=object_key,
            UploadId=mpu['UploadId'],
            MultipartUpload={"Parts": parts}
        )

    except ClientError as e:
        print(f"Error: {e}")
        s3_client.abort_multipart_upload(Bucket=bucket_name, Key=object_key, UploadId=mpu['UploadId'])

write

python import boto3 import csv import io from botocore.exceptions import ClientError  def write(config, configured_catalog, input_messages):     s3_client = boto3.client(         's3',         aws_access_key_id=config['aws_access_key_id'],         aws_secret_access_key=config['aws_secret_access_key']     )     bucket_name = config['bucket_name']     object_key = config['object_key']      # Khởi tạo multipart upload     try:         mpu = s3_client.create_multipart_upload(Bucket=bucket_name, Key=object_key)         parts = []         part_number = 1         buffer = io.StringIO()         csv_writer = csv.writer(buffer)          for msg in input_messages:             if msg.type == Type.RECORD:                 csv_writer.writerow([msg.record.stream, msg.record.data, msg.record.emitted_at])                                  # Nếu buffer đạt kích thước nhất định, upload một phần                 if buffer.tell() > 5 * 1024 * 1024:  # 5MB                     buffer.seek(0)                     part = s3_client.upload_part(                         Bucket=bucket_name,                         Key=object_key,                         PartNumber=part_number,                         UploadId=mpu['UploadId'],                         Body=buffer.getvalue().encode()                     )                     parts.append({"PartNumber": part_number, "ETag": part['ETag']})                     part_number += 1                     buffer = io.StringIO()                     csv_writer = csv.writer(buffer)          # Upload phần cuối cùng nếu còn dữ liệu trong buffer         if buffer.tell() > 0:             buffer.seek(0)             part = s3_client.upload_part(                 Bucket=bucket_name,                 Key=object_key,                 PartNumber=part_number,                 UploadId=mpu['UploadId'],                 Body=buffer.getvalue().encode()             )             parts.append({"PartNumber": part_number, "ETag": part['ETag']})          # Hoàn thành multipart upload         s3_client.complete_multipart_upload(             Bucket=bucket_name,             Key=object_key,             UploadId=mpu['UploadId'],             MultipartUpload={"Parts": parts}         )      except ClientError as e:         print(f"Error: {e}")         s3_client.abort_multipart_upload(Bucket=bucket_name, Key=object_key, UploadId=mpu['UploadId'])

check

python
import boto3
from botocore.exceptions import ClientError
from airbyte_cdk.models import AirbyteConnectionStatus, Status

def check(self, logger, config):
    """
    Kiểm tra xem cấu hình có cho phép kết nối thành công đến S3 bucket hay không.
    
    :param logger: Đối tượng logger để ghi log
    :param config: Cấu hình được cung cấp bởi người dùng
    :return: AirbyteConnectionStatus cho biết kết nối thành công hay thất bại
    """
    try:
        # Tạo client S3 với thông tin xác thực từ config
        s3_client = boto3.client(
            's3',
            aws_access_key_id=config['aws_access_key_id'],
            aws_secret_access_key=config['aws_secret_access_key'],
            region_name=config.get('region_name')  # region_name có thể không bắt buộc
        )

        # Thử liệt kê các objects trong bucket để kiểm tra quyền truy cập
        s3_client.list_objects_v2(Bucket=config['bucket_name'], MaxKeys=1)

        # Nếu không có lỗi, kết nối thành công
        return AirbyteConnectionStatus(status=Status.SUCCEEDED)

    except ClientError as e:
        # Xử lý các lỗi cụ thể
        error_code = e.response['Error']['Code']
        error_message = e.response['Error']['Message']
        
        if error_code == 'NoSuchBucket':
            message = f"Bucket '{config['bucket_name']}' không tồn tại."
        elif error_code in ['AccessDenied', 'InvalidAccessKeyId', 'SignatureDoesNotMatch']:
            message = "Lỗi xác thực. Vui lòng kiểm tra AWS credentials của bạn."
        else:
            message = f"Lỗi khi kết nối đến S3: {error_message}"

        logger.error(message)
        return AirbyteConnectionStatus(status=Status.FAILED, message=message)

    except Exception as e:
        # Xử lý các lỗi không mong đợi khác
        logger.error(f"Lỗi không mong đợi: {str(e)}")
        return AirbyteConnectionStatus(status=Status.FAILED, message=f"Lỗi không mong đợi: {str(e)}")

check

python import boto3 from botocore.exceptions import ClientError from airbyte_cdk.models import AirbyteConnectionStatus, Status  def check(self, logger, config):     """     Kiểm tra xem cấu hình có cho phép kết nối thành công đến S3 bucket hay không.          :param logger: Đối tượng logger để ghi log     :param config: Cấu hình được cung cấp bởi người dùng     :return: AirbyteConnectionStatus cho biết kết nối thành công hay thất bại     """     try:         # Tạo client S3 với thông tin xác thực từ config         s3_client = boto3.client(             's3',             aws_access_key_id=config['aws_access_key_id'],             aws_secret_access_key=config['aws_secret_access_key'],             region_name=config.get('region_name')  # region_name có thể không bắt buộc         )          # Thử liệt kê các objects trong bucket để kiểm tra quyền truy cập         s3_client.list_objects_v2(Bucket=config['bucket_name'], MaxKeys=1)          # Nếu không có lỗi, kết nối thành công         return AirbyteConnectionStatus(status=Status.SUCCEEDED)      except ClientError as e:         # Xử lý các lỗi cụ thể         error_code = e.response['Error']['Code']         error_message = e.response['Error']['Message']                  if error_code == 'NoSuchBucket':             message = f"Bucket '{config['bucket_name']}' không tồn tại."         elif error_code in ['AccessDenied', 'InvalidAccessKeyId', 'SignatureDoesNotMatch']:             message = "Lỗi xác thực. Vui lòng kiểm tra AWS credentials của bạn."         else:             message = f"Lỗi khi kết nối đến S3: {error_message}"          logger.error(message)         return AirbyteConnectionStatus(status=Status.FAILED, message=message)      except Exception as e:         # Xử lý các lỗi không mong đợi khác         logger.error(f"Lỗi không mong đợi: {str(e)}")         return AirbyteConnectionStatus(status=Status.FAILED, message=f"Lỗi không mong đợi: {str(e)}")

{
  "type": "LOG",
  "log": {
    "level": "FATAL",
    "message": "'AirbyteConnectionStatus' object is not iterable\nTraceback (most recent call last):\n  File \"/mnt/d/dft/airbyte/airbyte-integrations/connectors/destination-destination-s3-v3/main.py\", line 11, in <module>\n    DestinationDestinationS3V3().run(sys.argv[1:])\n  File \"/home/hungpm/.local/lib/python3.10/site-packages/airbyte_cdk/destinations/destination.py\", line 119, in run\n    for message in output_messages:\n  File \"/home/hungpm/.local/lib/python3.10/site-packages/airbyte_cdk/destinations/destination.py\", line 113, in run_cmd\n    yield from self._run_write(config=config, configured_catalog_path=parsed_args.catalog, input_stream=wrapped_stdin)\n  File \"/home/hungpm/.local/lib/python3.10/site-packages/airbyte_cdk/destinations/destination.py\", line 49, in _run_write\n    yield from self.write(config=config, configured_catalog=catalog, input_messages=input_messages)\nTypeError: 'AirbyteConnectionStatus' object is not iterable"
  }
}

write

write

AirbyteConnectionStatus

write

AirbyteConnectionStatus

write

python def write(self, config: Mapping[str, Any], configured_catalog: ConfiguredAirbyteCatalog, input_messages: Iterable[AirbyteMessage]) -> Iterable[AirbyteMessage]:     # Xử lý logic ghi dữ liệu vào S3     for message in input_messages:         if message.type == Type.RECORD:             # Xử lý record             pass         elif message.type == Type.STATE:             # Yield state message             yield message          # Yield các log messages nếu cần     yield AirbyteMessage(type=Type.LOG, log=AirbyteLogMessage(level=<http://Level.INFO|Level.INFO>, message="Write completed successfully"))

check

write

check

AirbyteConnectionStatus

write

AirbyteMessage

check

write

yield

python def write(self, config: Mapping[str, Any], configured_catalog: ConfiguredAirbyteCatalog, input_messages: Iterable[AirbyteMessage]) -> Iterable[AirbyteMessage]:     <http://self.logger.info|self.logger.info>("Starting write method")     # Rest of your write method...

write