DynamoDB // Lambda //记录不一致

如何解决DynamoDB // Lambda //记录不一致

我们基于事件将数据从应用程序推送到DynamoDB。我们编写了一个lambda函数，该函数映射到此DynamoDB表。我们在应用程序中更新的记录已成功发送到DyanamoDB。但是，lambda通常会减少记录的数量。

如果我们的应用程序中更新了3条记录，则所有3条记录都将推送到Dynamo，但是lambda有时只给出2条或1条。

我们在lambda函数中构建了一个csv文件，因此该csv文件的记录数量也减少了。

我可以共享一些代码段，但是我不知道这与Lambda配置或我们用来将数据推送到Dynamo的Dynamo API有什么关系吗？

以下是Lambda代码：

public class ProductLambda implements RequestHandler<DynamodbEvent,String> {
    private static final String M_NUM = "number";
    private static final String FT_TABLE = "table";
    private AmazonS3 s3 = 
    AmazonS3ClientBuilder.standard().withRegion(Regions.US_EAST_1).build();
    private static final String S3_BUCKET = "bucket";
    private static final String S3_FOLDER = "folder";

public ProductLambda() {
}

@Override
public String handleRequest(DynamodbEvent event,Context context) {
    String[] appcol = new String[] {"COL1","COL2","COL3","COL4","COL1","COL4"};
    String[] fttcol = new String[] {"COL1","COL4"};
    
    createFolder(S3_BUCKET,S3_FOLDER,s3);
    List<DynamodbStreamRecord> records = event.getRecords();
    try {
        List<LinkedHashMap<String,AttributeValue>> changedRecords = new ArrayList<LinkedHashMap<String,AttributeValue>>();
        String ddbTable = "";
        for (DynamodbStreamRecord record : records) {
            String modelNumber = "";
            String ddbARN = record.getEventSourceARN();
            if (ddbARN != null) {
                ddbTable = ddbARN.split("/")[1];
            }
            if (record != null) {
                Map<String,AttributeValue> changedRec = record.getDynamodb().getNewImage();
                if (changedRec != null) {
                    if (ddbTable.equalsIgnoreCase("table1")) {
                        LinkedHashMap<String,AttributeValue> staticColumnRecord = new LinkedHashMap<String,AttributeValue>();
                        List<String> list1 = new ArrayList<String>(Arrays.asList(appcol));
                        for (String key : list1) {
                            AttributeValue value = changedRec.get(key);
                            if (value != null) {
                                staticColumnRecord.put(key,value);
                            } else {
                                staticColumnRecord.put(key,new AttributeValue().withS("null"));
                            }
                        }
                        changedRecords.add(staticColumnRecord);
                    }
                    if (ddbTable.equalsIgnoreCase("table2")) {
                        LinkedHashMap<String,AttributeValue>();
                        List<String> list2 = new ArrayList<String>(Arrays.asList(fttcol));
                        for (String key : list2) {
                            AttributeValue value = changedRec.get(key);
                            if(key.equals("ft_mod")){
                                modelNumber = value.getS();
                            }
                            if (value != null) {
                                staticColumnRecord.put(key,new AttributeValue().withS("null"));
                            }
                        }
                        staticColumnRecord = getModelDetails(staticColumnRecord,modelNumber);
                        changedRecords.add(staticColumnRecord);
                    } 
                }
            }
        }
        context.getLogger().log("[DEBUG] - Total Records Changed: " + changedRecords.size());
        buildAndUploadCSV(changedRecords,context,ddbTable);
    } catch (AmazonServiceException e) {
        context.getLogger().log("[ERROR] - AmazonServiceException " + e);
    } catch (AmazonClientException e) {
        context.getLogger().log("[ERROR] - AmazonClientException " + e);
    } catch (IOException e) {
        context.getLogger().log("[ERROR] - IOException " + e);
    }
    context.getLogger().log("[DEBUG] - Total Records are: " + records.size());
    return "Total records are: " + records.size();
}

/*
 * 
 * @param staticColumnRecord
 * @param modelNumber
 * @param context
 * @return
 */
private LinkedHashMap<String,AttributeValue> getModelDetails(LinkedHashMap<String,AttributeValue> staticColumnRecord,String modelNumber) {

    String[] modcol = new String[] { "COL1","COL4" };
    if (!modelNumber.isEmpty()) {
        AmazonDynamoDB dynamoClient = AmazonDynamoDBClientBuilder.standard().build();
        GetItemRequest getItemRequest = new GetItemRequest().withTableName(FT_TABLE)
                .addKeyEntry(M_NUM,new AttributeValue().withS(modelNumber));
        Map<String,AttributeValue> responseItem = dynamoClient.getItem(getItemRequest).getItem();
        List<String> sf_mode_col = new ArrayList<String>(Arrays.asList(modcol));
        if (responseItem != null) {
            for (String key : sf_mode_col) {
                AttributeValue val = responseItem.get(key);
                if (val != null) {
                    staticColumnRecord.put(key,val);
                } else {
                    staticColumnRecord.put(key,new AttributeValue().withS("null"));
                }
            }
        }
    }
    return staticColumnRecord;
}

/*
 * Creates a folder in the S3 bucket.
 */
private static void createFolder(String bucketName,String folderName,AmazonS3 client) {
    ObjectMetadata metadata = new ObjectMetadata();
    metadata.setContentLength(0);
    InputStream emptyContent = new ByteArrayInputStream(new byte[0]);
    PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName,folderName,emptyContent,metadata);
    PutObjectResult result = client.putObject(putObjectRequest);
}

/*
 * Build CSV file of changed data and upload to S3 bucket.
 */
private void buildAndUploadCSV(List<LinkedHashMap<String,AttributeValue>> changedRecords,Context context,String tableName)
        throws IOException {
    Calendar calendar = Calendar.getInstance();
    SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMddHHmmss");
    String outputName = tableName + "_" + formatter.format(calendar.getTime()) + ".csv";
    List<String> headers = changedRecords.stream().flatMap(map -> map.keySet().stream()).distinct()
            .collect(Collectors.toList());
    Path filePath = Paths.get("/tmp",outputName);
    try (OutputStreamWriter bw = new OutputStreamWriter(new FileOutputStream(filePath.toFile()),StandardCharsets.UTF_8)) 
    {
        StringBuffer headerContent = new StringBuffer();
        for (String string : headers) {
            headerContent.append(string);
            headerContent.append(",");
        }
        if (headerContent.length() > 0  && headerContent.charAt(headerContent.length() - 1) == ',')
            headerContent.setLength( headerContent.length() - 1 );

        StringBuffer strBuilder = new StringBuffer();
        for (LinkedHashMap<String,AttributeValue> lmap : changedRecords) {
            StringBuilder stringBuilder = new StringBuilder("");
            String sep = "";
            for (Entry<String,AttributeValue> string2 : lmap.entrySet()) {
                String value = string2.getValue().getS();
                stringBuilder.append(sep).append("\"").append(value).append("\"");
                sep = ",";
            }
            if (!stringBuilder.toString().isEmpty()) {
                strBuilder.append(stringBuilder).append(System.getProperty("line.separator"));
            }
        }
        headerContent.append("\n");
        headerContent.append(strBuilder);
        ByteArrayInputStream contentsAsStream = new ByteArrayInputStream(headerContent.toString().getBytes());
        ObjectMetadata md = new ObjectMetadata();   
        md.setContentLength((long)headerContent.toString().getBytes().length);   
        md.setContentType("text/csv");
        PutObjectRequest request = new PutObjectRequest("bucket","folder/" + outputName,contentsAsStream,md);
        s3.putObject(request);
        bw.flush();
        bw.close();
    } catch (IOException e) {
        context.getLogger().log("[ERROR] - IOException " + e);
    }
}

}