AWS Kinesis Stream写入RDS PostgreSQL数据不全及配置疑问咨询

Hey there! Let's dig into why you're missing those 18,000 records and whether setting BatchSize=1 will get you full coverage. I've worked through similar AWS bulk import scenarios, so here's what I'd check and recommend:

Why You're Missing Records #

There are a few likely culprits here—let's break them down:

• Uncaught Failures in the Kinesis-to-PostgreSQL Lambda
  If your second Lambda hits an error (like a PostgreSQL constraint violation, connection timeout, or malformed record), it might retry the batch a few times, but if the error persists, AWS will drop those records (unless you have a dead-letter queue configured). Check your CloudWatch Logs for this Lambda—look for exceptions like SQLException, BatchUpdateException, or even Kinesis processing errors. A single bad record could take down an entire batch, leaving you with partial imports.

• Throttling or Unhandled Errors in the S3-to-Kinesis Lambda
  Kinesis Data Streams have per-shard limits (1MB/s or 1,000 records/s by default). If your first Lambda tries to write faster than this, you'll hit ProvisionedThroughputExceededException. If you don't have retry logic for this, those failed PutRecord calls mean records never make it to Kinesis. Also, double-check your CSV reading code—are you properly handling edge cases like line breaks inside quoted fields? Using a proper CSV parser (like OpenCSV) instead of raw BufferedReader can prevent accidental line splitting that skips records.

• Kinesis Retention & Processing Lag
  Kinesis retains records for 24 hours by default, but if your second Lambda is backed up (e.g., due to slow database writes), it might not catch up before records expire. Though 50k records shouldn't hit this unless processing is extremely slow, it's worth verifying your Lambda's invocation metrics in CloudWatch to see if there's backlog.

Will Setting BatchSize=1 Guarantee Full Import? #

Sort of—but it's more of a band-aid than a full fix. Here's the lowdown:

• Setting BatchSize=1 ensures that a single bad record won't take down an entire batch. If one record fails, only that one is retried (or dropped, if no DLQ is set up), instead of losing all records in the batch.
• However, this won't fix underlying issues like throttling in the first Lambda, or unhandled database errors. You'll still need to handle retries for Kinesis writes and database operations, plus capture failed records somewhere (like an SQS dead-letter queue) so you can debug and reprocess them.
• Also, keep in mind that BatchSize=1 will increase Lambda invocation count, which can bump up costs and trigger concurrency limits if you're processing thousands of records quickly.

Recommended Fixes & Configuration Tweaks #

Let's get your import working reliably:

1. Fix the S3-to-Kinesis Lambda #

• Add Retry Logic for Kinesis Writes: Use the AWS SDK's built-in retry policies (or implement exponential backoff) to handle ProvisionedThroughputExceededException. For Java, you can configure the KinesisClient with a RetryPolicy that retries on this exception.
• Validate CSV Reading: Swap raw line reading for a CSV library like OpenCSV to correctly parse fields with embedded newlines or quotes. Log the number of lines you read from S3 vs. the number of PutRecord calls you make—this will tell you if records are getting lost before they even hit Kinesis.
• Add CloudWatch Metrics: Track custom metrics for lines read vs. lines written to Kinesis to quickly spot discrepancies.

2. Fix the Kinesis-to-PostgreSQL Lambda #

• Configure a Dead-Letter Queue (DLQ): Attach an SQS queue as a DLQ to your Lambda. Any records that fail processing after all retries will be sent here, so you can inspect them later instead of losing them.
• Handle Database Errors Gracefully: Catch SQLException and differentiate between retriable errors (e.g., connection timeouts) and non-retriable errors (e.g., duplicate primary keys). Retry retriable errors with backoff, and send non-retriable errors straight to the DLQ.
• Adjust BatchSize Strategically: If you don't want to use BatchSize=1, try a smaller batch size (like 10 or 50) to balance speed and fault tolerance. This reduces the impact of a single bad record while keeping invocation counts manageable.

3. Optimize Kinesis Configuration #

• Check Shard Count: If your first Lambda is hitting throttling, increase the number of shards in your Kinesis stream. Each shard adds more throughput capacity.
• Enable Enhanced Fan-Out: This lets your Lambda read directly from Kinesis without competing with other consumers, reducing processing lag.

4. Consider a Simpler Bulk Import Approach (For One-Time Jobs) #

Since this is a one-time import, you might not need the Lambda/Kinesis pipeline at all. RDS PostgreSQL supports direct COPY from S3, which is way faster and more reliable for bulk loads:

COPY your_target_table 
FROM 's3://your-bucket/path/to/your/file.csv'
IAM_ROLE 'arn:aws:iam::your-account-id:role/your-s3-access-role'
CSV HEADER;

This command pulls the CSV directly from S3 into your PostgreSQL table in one go, no middleman required. Just make sure your RDS instance has permission to access the S3 bucket via an IAM role.

内容的提问来源于stack exchange，提问作者ShaiNe Ram