nifiでGetDynamoDBプロセッサのJson Document属性を設定する方法

Question

GetDynamoDBプロセッサをnifiで使用してDynamoDBからデータを取得しようとしていますが、Json Document属性以外のすべての必須フィールドを提供していますフィールド。

入力データ：

{ 
"ProductCatalog": [ 
    { 
     "PutRequest": { 
      "Item": { 
       "Id": { 
        "N": "101" 
       }, 
       "Title": { 
        "S": "Book 101 Title" 
       }, 
       "ISBN": { 
        "S": "111-1111111111" 
       }, 
       "Authors": { 
        "L": [ 
         { 
          "S": "Author1" 
         } 
        ] 
       }, 
       "Price": { 
        "N": "2" 
       }, 
       "Dimensions": { 
        "S": "8.5 x 11.0 x 0.5" 
       }, 
       "PageCount": { 
        "N": "500" 
       }, 
       "InPublication": { 
        "BOOL": true 
       }, 
       "ProductCategory": { 
        "S": "Book" 
       } 
      } 
     } 
    }] 
    } 

いずれかを理解することができます。

Answer 1

JSONドキュメントをDynamoDBに保存すると、保存するJSONドキュメントは、というDynamoDBアイテムというラッパーオブジェクト内にネストされたオブジェクトとして配置され、ドキュメントに関するメタデータが格納されます。 Json Document attributeでは、ネストされたJSONドキュメントのDynamoDBアイテムにキーを指定できるため、トップレベルのDynamoDBアイテムから抽出することができます。 source code for the GetDynamoDB NiFi processorを見ると

は値（jsonDocument）は、AWS DynamoDBのSDKを呼び出し、指定DynamoDBの項目属性にJSONを抽出し、その内容のためNiFi FlowFileを作成するために使用される方法を示しています。

BatchGetItemOutcome result = dynamoDB.batchGetItem(tableKeysAndAttributes); 

// Handle processed items and get the json document 
List<Item> items = result.getTableItems().get(table); 
for (Item item : items) { 
    ItemKeys itemKeys = new ItemKeys(item.get(hashKeyName), item.get(rangeKeyName)); 
    FlowFile flowFile = keysToFlowFileMap.get(itemKeys); 

    if (item.get(jsonDocument) != null) { 
     ByteArrayInputStream bais = new ByteArrayInputStream(item.getJSON(jsonDocument).getBytes()); 
     flowFile = session.importFrom(bais, flowFile); 
    } 

    session.transfer(flowFile,REL_SUCCESS); 
    keysToFlowFileMap.remove(itemKeys); 
} 

お役に立てれば！