#aws boto3 dynamodb | Explore Tumblr posts and blogs

globalmediacampaign · 4 years ago

Text

Integrate your Amazon DynamoDB table with machine learning for sentiment analysis

Amazon DynamoDB is a non-relational database that delivers reliable performance at any scale. It’s a fully managed, multi-Region, multi-active database that provides consistent single-digit millisecond latency and offers built-in security, backup and restore, and in-memory caching. DynamoDB offers a serverless and event-driven architecture, which enables you to use other AWS services to extend DynamoDB capability. DynamoDB provides this capability using Kinesis Data Streams for DynamoDB and DynamoDB Streams with AWS Lambda. When you enable DynamoDB Streams on a DynamoDB table, it captures a time-ordered sequence of item-level modifications in the table and stores this information in a change log for up to 24 hours. Downstream AWS services can access these change logs and view the data items as they appeared before and after they were modified, in near-real time, using a Lambda function. This allows the DynamoDB table to integrate functionally for additional use cases like machine learning (ML), ad hoc queries, full text search, event alerting, and more, such as the following: Processing DynamoDB data with Apache Hive on Amazon EMR for data warehousing use cases Configuring AWS credentials using Amazon Cognito to authenticate access to DynamoDB tables for mobile application Using Amazon API Gateway and Lambda with DynamoDB for a front-end serverless architecture In this post, I show you how to integrate ML with Amazon DynamoDB using Amazon Comprehend to analyze sentiments on incoming product reviews. Serverless and event-driven architecture When you perform create, read, update, and delete (CRUD) operations on a DynamoDB table, DynamoDB Streams keep a 24-hour change log of all CRUD operations. The stream offers you four options of log attributes. For more information, see Enabling a Stream. Lambda functions process each incoming log event, extract the necessary information required for downstream services, and invoke said services, such as in the following situations: A function can loop back to a DynamoDB table to create or update the aggregate item, like a summary item or average of statistics in real time. You can use a function to send specific attributes to Amazon Elasticsearch Service (Amazon ES) for full text search use cases. For historical analysis or ad hoc queries, the function sends the updates to Amazon Simple Storage Service (Amazon S3) in optimized format through Amazon Kinesis Data Firehose, where you can run ad hoc queries or analytics using Amazon Athena You can use the recently announced option to export DynamoDB table data to Amazon S3 for the initial setup of data for analysis; however, you can’t use it for continuous read requirements. Additionally, using Kinesis Data Firehose for data export offers optimized Apache Parquet If you want to be updated on a specific event on the DynamoDB table like the deletion of a critical record, Lambda can notify you through Amazon Simple Notification Service (Amazon SNS). The following diagram illustrates this event-driven architecture. Using ML and analytics with the DynamoDB solution You can configure Amazon review analysis and analytics by running AWS CloudFormation on your own account. This solution uses a serverless event-driven architecture with automated steps. The solution checks the sentiment on the incoming Amazon product review, creates a product review summary based on the sentiment, and keeps updates in optimized format for future ad hoc queries and analytics. The following diagram illustrates this architecture. For this post, the solution uses the incoming Amazon product reviews by putting a review object in the S3 bucket in the form of JSON records. The S3 bucket publishes the s3:ObjectCreated:put event to Lambda by invoking the Lambda function, as specified in the bucket notification configuration. Lambda is meant for quick processing, so the solution limits the size of the product review object on Amazon S3 to smaller than 5 KB. If you want to run this solution in your account, follow these steps: Download the following artifacts: CloudFormation template – Part 1 CloudFormation template – Part 2 Sample gz file for testing – test.json.gz Deploy the CloudFormation Part 1 stack. Check the CloudFormation documentation for instructions. Once Part 1 stack deployment is complete, deploy the Part 2 stack. Once Part 2 stack deployment is complete, upload the sample gz file to the S3 bucket (-amazonreviewsbucket-*) with Server-side encryption option. Check Amazon S3 documentation for instructions. The CloudFormation Part 1 stack creates the following resources: A Glue database and table for cataloging, Two Lambda functions and associated permissions, Required roles and policies, CloudWatch LogStream and LogGroup, Kinesis Firehose for sending streams data to S3 for analysis, Two S3 buckets for incoming reviews and parquet output for analysis, DynamoDB table and associated Streams. The CloudFormation Part 2 stack imports the resources created and setup a managed policy. You can run the stacks on all regions where these services are supported. I tested it on US regions, North Virginia, Oregon and Ohio. Please note running this solution in your account will incur costs. The Lambda function amazon-review-processing runs and assumes the AWS Identity and Access Management (IAM) role created by AWS CloudFormation. The function reads the Amazon S3 events it receives as a parameter, determines where the review object is, reads the review object, and processes the records in the review object. The function breaks the incoming record into multiple records and DynamoDB items: a review item and product item (if it doesn’t already exist). This allows you to retrieve an individual item based on the partition key and sort key. If you want all the product items, you can query items based on just the product partition key. The following screenshot shows product summary item. The following screenshot shows review items associated with the product. The following Lambda code add the review data to DynamoDB table when review file is uploaded to S3 bucket. import io import os import gzip import json import boto3 def get_records(session, bucket, key): """ Generator for the bucket and key names of each CloudTrail log file contained in the event sent to this function from S3. (usually only one but this ensures we process them all). :param event: S3:ObjectCreated:Put notification event :return: yields bucket and key names """ s3 = session.client('s3') response = s3.get_object(Bucket=bucket, Key=key) with io.BytesIO(response['Body'].read()) as obj: with gzip.GzipFile(fileobj=obj) as logfile: records = json.load(logfile) return records def handler(event, context): """ Checks for API calls with RunInstances, TerminateInstances, and DeleteDBInstance in CloudTrail. if found, send specific records to SQS for processing :return: 200, success if records process successfully """ session = boto3.session.Session() dynamodb = boto3.resource("dynamodb", region_name='us-east-1') table = dynamodb.Table('AmazonReviews') # Get the S3 bucket and key for each log file contained in the event for event_record in event['Records']: try: bucket = event_record['s3']['bucket']['name'] key = event_record['s3']['object']['key'] print('Loading Amazon Reviews file s3://{}/{}'.format(bucket, key)) records = get_records(session, bucket, key) print('Number of records in log file: {}'.format(len(records))) for record in records: response = table.get_item(Key={'pk': record['product_id'], 'sk': '2099-12-31#PRODUCTSUMMARY'}) if 'Items' not in response: table.put_item( Item={ 'pk': record['product_id'], 'sk': '2099-12-31#PRODUCTSUMMARY', 'marketplace': record['marketplace'], 'product_parent': record['product_parent'], 'product_title': record['product_title'], 'product_category': record['product_category'], } ) table.put_item( Item={ 'pk': record['product_id'], 'sk': record['review_date'] + '#' + record['review_id'], 'customer_id': record['customer_id'], 'star_rating': record['star_rating'], 'helpful_votes': record['helpful_votes'], 'total_votes': record['total_votes'], 'vine': record['vine'], 'verified_purchase': record['verified_purchase'], 'review_headline': record['review_headline'], 'review_body': record['review_body'] } ) except Exception as e: print (e) return {'Exception status': e} else: print("records processed successfully!!") return { 'statusCode': 200, 'body': json.dumps('records inserted successfully to DynamoDB!!') } After the review records are added to the DynamoDB table, the items enter the DynamoDB stream in real time with new and old images. The amazon_reviews_summary Lambda function captures the stream records from the stream and processes them one by one. This Lambda function has multiple responsibilities: Capture review text from the stream record and call Amazon Comprehend for sentimental analysis. Amazon Comprehend limits review strings to fewer than 5,000 characters, so the code truncates review text to 4,999 characters before calling the Amazon Comprehend sentiment API. Add the sentiment response for the review record and create a product review summary record with sentiment counts. Flatten the DynamoDB streaming JSON logs and add the record to the Kinesis Data Firehose delivery stream. Invoke the Firehose delivery stream put_record API for putting updates in the S3 bucket. The following Lambda code processes the incoming DynamoDB Streams for sentiment analysis and save it to S3 for analytics. import json import boto3 import os def convert_file(f): out = {} def convert(element, name=''): if type(element) is dict: for sub in element: convert(element[sub], name + sub + '_') elif type(element) is list: ctr = 0 for sub in element: convert(sub, name + str(ctr) + '_') ctr += 1 else: out[name[:-1]] = element convert(f) return out def handler(event, context): cmphd = boto3.client(service_name='comprehend', region_name='us-east-1') fh = boto3.client('firehose') ddb = boto3.resource('dynamodb', region_name='us-east-1') dt=ddb.Table('AmazonReviews') FIREHOSE_URL = os.environ['FIREHOSE_URL'] for rec in event['Records']: if (rec['eventName'] == 'INSERT' and ('review_body' in rec['dynamodb']['NewImage'])): convt=convert_file(rec) response = fh.put_record( DeliveryStreamName=FIREHOSE_URL, Record={'Data': json.dumps(convt)} ) review_body=rec['dynamodb']['NewImage']['review_body']['S'] review_body=review_body[:4999] pk=rec['dynamodb']['Keys']['pk']['S'] sk=rec['dynamodb']['Keys']['sk']['S'] res=cmphd.detect_sentiment(Text=review_body, LanguageCode='en') st=res['Sentiment'] try: d_response = dt.put_item( Item={'pk': pk, 'sk': sk + '#' + 'SENTIMENT', 'sentiment': st} ) if st == "POSITIVE": d_s_response = dt.update_item( Key={'pk': pk,'sk': '2099-12-31#REVIEWSUMMARY'}, UpdateExpression="set positive_sentiment_count= if_not_exists(positive_sentiment_count, :start) + :inc",ExpressionAttributeValues={':inc': 1,':start': 0},ReturnValues="UPDATED_NEW" ) elif st == "NEGATIVE": d_s_response = dt.update_item( Key={'pk': pk,'sk': '2099-12-31#REVIEWSUMMARY'}, UpdateExpression="set negative_sentiment_count= if_not_exists(negative_sentiment_count, :start) + :inc",ExpressionAttributeValues={':inc': 1,':start': 0},ReturnValues="UPDATED_NEW" ) elif st == "NEUTRAL": d_s_response = dt.update_item( Key={'pk': pk,'sk': '2099-12-31#REVIEWSUMMARY'}, UpdateExpression="set neutral_sentiment_count= if_not_exists(neutral_sentiment_count, :start) + :inc",ExpressionAttributeValues={':inc': 1,':start': 0},ReturnValues="UPDATED_NEW" ) elif st == "MIXED": d_s_response = dt.update_item( Key={'pk': pk,'sk': '2099-12-31#REVIEWSUMMARY'}, UpdateExpression="set mixed_sentiment_count= if_not_exists(mixed_sentiment_count, :start) + :inc",ExpressionAttributeValues={':inc': 1,':start': 0},ReturnValues="UPDATED_NEW" ) else: print("No sentiment value: " + st) except Exception as e: return {'Exception status': e} else: print("record processed successfully") The amazon_reviews_summary Lambda function calls the Amazon Comprehend detect_sentiment API with the review text, and Amazon Comprehend returns one of the four sentiments: positive, negative, neutral, or mixed. If you want more granular sentiments, you can replace Amazon Comprehend with your own ML model. The amazon_reviews_summary Lambda function calls the DynamoDB table to update the sentiment response on the review item and create product review summary items (or update them if they already exist). The following screenshot shows sentiment item associated with review. The following screenshot shows review summary item. The amazon_reviews_summary Lambda function calls the Firehose delivery stream to flatten the DynamoDB streaming JSON payload. The delivery stream converts the record format to Parquet and adds these records to the S3 bucket. Kinesis Firehose is configured with Record format conversion enabled and Output format as Apache Parquet. You can now catalog the S3 bucket on Athena and run ad hoc queries. See the following query and result: SELECT * FROM "amazon_review"."amazon_reviews" where dynamodb_keys_pk_s ='B00004U2JW' eventid,eventname,eventversion,eventsource,awsregion,dynamodb_approximatecreationdatetime,dynamodb_keys_sk_s,dynamodb_keys_pk_s,dynamodb_newimage_total_votes_n,dynamodb_newimage_star_rating_n,dynamodb_newimage_sk_s,dynamodb_newimage_verified_purchase_s,dynamodb_newimage_pk_s,dynamodb_newimage_customer_id_n,dynamodb_newimage_helpful_votes_n,dynamodb_newimage_vine_s,dynamodb_newimage_review_body_s,dynamodb_newimage_review_headline_s,dynamodb_sequencenumber,dynamodb_sizebytes,dynamodb_streamviewtype,eventsourcearn 35f3c7ce3083aa72eb6b39cd56f3681d,INSERT,1.1,aws:dynamodb,us-east-1,1598908834,2000-09-07#R3MA96SRZSSBER,B00004U2JW,27,4,2000-09-07#R3MA96SRZSSBER,N,B00004U2JW,48308880,5,N,"cool camera with functions whitch i do need, but I cant buy her /and others from my wish list/ because I am not living in U.S. Amazon hurry, she is one of my favourite...",cool but to long way to get her to me,8.53782E+25,393,NEW_AND_OLD_IMAGES,arn:aws:dynamodb:us-east-1:080931190378:table/amazon-reviews-v1/stream/2020-08-14T06:28:33.219 Cleaning up To avoid incurring future costs, delete the resources you created with the CloudFormation stack: Delete the contents of the review bucket and Parquet analytics bucket. Delete the Part 2 CloudFormation stack first and then delete the Part 1 CloudFormation stack. Conclusion In this post, I showed how you can integrate your DynamoDB table with Amazon Comprehend for sentiment analysis and Athena for ad hoc queries across historical activities. You can use your own ML model instead of Amazon Comprehend if you want more granular sentiment analysis. You can also use other AWS services to integrate with DynamoDB by using DynamoDB Streams to extend DynamoDB features. I welcome your feedback, suggestions or questions on this post, please leave me comments below. About the author Utsav Joshi is a Technical Account Manager at AWS. He lives in New Jersey and enjoys working with AWS customers to solve architectural, operational, and cost optimization challenges. In his spare time, he enjoys traveling, road trips, and playing with his kids. https://aws.amazon.com/blogs/database/integrate-your-amazon-dynamodb-table-with-machine-learning-for-sentiment-analysis/

0 notes

awsexchage · 6 years ago

Photo

Amazon Managed BlockchainでHyperledger Fabricのブロックチェーンネットワークをさくっと構築するAWS CloudFormationのテンプレートを作ってみた(解説編) https://ift.tt/2M19PhQ

Amazon Managed Blockchain(AMB)でブロックチェーンネットワークを構築するのが手間になったので、AWS CloudFormation(CFn)を利用して構築できるようにしてみました。使い方は下記をご参考ください。

Amazon Managed BlockchainでHyperledger Fabricのブロックチェーンネットワークをさくっと構築するAWS CloudFormationのテンプレートを作ってみた(使い方編) – Qiita https://cloudpack.media/48077

Cfnのテンプレートを作成するのにいくつかハマったりしたので解説がてらまとめてみます。テンプレートはGitHubにアップしています。

kai-kou/amazon-managed-blockchain-cfn-template https://github.com/kai-kou/amazon-managed-blockchain-cfn-template

CFnがAMBリソースに対応していない

AMBのネットワークやメンバーなどのリソースがCFnでサポートされていません。(2019/07/03時点)

AWS Resource and Property Types Reference – AWS CloudFormation https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-template-resource-type-ref.html

Release History – AWS CloudFormation https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/ReleaseHistory.html

そのためAWS Lambda-backedカスタムリソースを利用してLambda関数でAWS SDKを利用してAMBのネットワークやメンバーを作成する必要があります。

詳細は下記が参考になります。

AWS SDK for Python(boto3)でAmazon Managed Blockchainのブロックチェーンネットワークを作成してみた – Qiita https://cloudpack.media/47241

AWS CloudFormationのLambda-backedカスタムリソースでリソースの更新・削除をする方法 – Qiita https://cloudpack.media/48205

リソースの作成順を制御する

AMBのPeerノードはネットワーク(メンバー)が利用可能になってから追加する必要があるため、CFnのDependsOn 属性でリソースの作成順を制御してネットワーク(→メンバー)→Peerノードの順にリソースを作成します。 ※ネットワークで最初のメンバーはネットワークと同時に作成する必要があります。

DependsOn 属性 – AWS CloudFormation https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/aws-attribute-dependson.html

AWS Lambda-backedカスタムリソースでリソース作成完了を待ち受ける

DependsOn 属性でリソースの作成順を制御することができたのですが、AMBのリソース(ネットワーク、Peerノード)をAWS CLIやAWS SDKで作成するとNetworkIdなどがすぐにレスポンスとして返ってきます。AMBだと「レスポンスが返ってきた=リソース作成できた」ではなく「レスポンスが返ってきた=リソース作成が開始された」となります。リソースが作成完了し利用可能になるには20分ほどかかります。

なので、CFnでネットワーク作成してすぐにPeerノードを作成しようとするとネットワークが利用可能になっておらずエラーになります。

そのためCFnでリソースの作成を待ち受けるのにAWS Lambda-backedカスタムリソースでどうにかする必要がありますが、AMBネットワークの作成には20分程度かかり、AWS Lmabdaのタイム��ウト(15分)を超えてしまうため、作成用のカスタムリソースだけでは待受けすることができません。そこで、待受用のカスタムリソースを定義することで、AWS Lambdaのタイムタウト(15分) x リトライ回数(3回)で45分まで待ち受けられるようにします。詳細は下記が参考になります。

AWS CloudFormationのLambda-Backedカスタムリソースでリソース作成を待ち受けできるようにする – Qiita https://cloudpack.media/48222

このリソース作成を待ち受ける実装とDependsOn属性を利用することでAMBのリソースが作成できるようになりました。下記はCFnのテンプレートから抜粋したリソース定義となります。

テンプレート抜粋

Resources: # ネットワーク作成用のカスタムリソース CreateBlockchainNetwork: Type: Custom::CustomResource # リソース作成待受用のカスタムリソース BlockchainNetwork: Type: Custom::CustomResource DependsOn: CreateBlockchainNetwork # リソース情報取得用のカスタムリソース BlockchainMember: Type: Custom::CustomResource DependsOn: BlockchainNetwork # Peerノード作成用のカスタムリソース CreateBlockchainPeerNode: Type: Custom::CustomResource DependsOn: BlockchainMember # リソース作成待受用のカスタムリソース BlockchainPeerNode: Type: Custom::CustomResource DependsOn: CreateBlockchainPeerNode

AWS Lambdaで利用できるAWS SDKを最新にする

AMBは2019/05/01にGAとなったサービスです。

New – Amazon Managed Blockchain – Create & Manage Scalable Blockchain Networks | AWS News Blog https://aws.amazon.com/jp/blogs/aws/new-amazon-managed-blockchain-create-manage-scalable-blockchain-networks

AWS Lambdaの関数(Python)で利用できるAWS SDK(boto3 1.9.42)だとAMBが対応していないバージョンとなるため、AWS Lambda Layersを利用して最新のAWS SDK(boto3 1.9.139 以上)を利用する必要があります。(2019/07/03時点)

boto3/CHANGELOG.rst at develop · boto/boto3 https://github.com/boto/boto3/blob/develop/CHANGELOG.rst#19139

api-change:managedblockchain: [botocore] Update managedblockchain client to latest version

エラー例

Lambda関数の実装

import json import boto3 def lambda_handler(event, context): print(boto3.__version__) client = boto3.client("managedblockchain") return { 'statusCode': 200, 'body': json.dumps('Hello from Lambda!') }

実行ログ

START RequestId: 1f574950-f61d-4d05-a2b6-a56e11eb2201 Version: $LATEST 1.9.42 [ERROR] UnknownServiceError: Unknown service: 'managedblockchain'. Valid service names are: acm, acm-pca, alexaforbusiness, apigateway, application-autoscaling, appstream, appsync, athena, autoscaling, autoscaling-plans, batch, budgets, ce, chime, cloud9, clouddirectory, cloudformation, cloudfront, cloudhsm, cloudhsmv2, cloudsearch, cloudsearchdomain, cloudtrail, cloudwatch, codebuild, codecommit, codedeploy, codepipeline, codestar, cognito-identity, cognito-idp, cognito-sync, comprehend, config, connect, cur, datapipeline, dax, devicefarm, directconnect, discovery, dlm, dms, ds, dynamodb, dynamodbstreams, ec2, ecr, ecs, efs, eks, elasticache, elasticbeanstalk, elastictranscoder, elb, elbv2, emr, es, events, firehose, fms, gamelift, glacier, glue, greengrass, guardduty, health, iam, importexport, inspector, iot, iot-data, iot-jobs-data, iot1click-devices, iot1click-projects, iotanalytics, kinesis, kinesis-video-archived-media, kinesis-video-media, kinesisanalytics, kinesisvideo, kms, lambda, lex-models, lex-runtime, lightsail, logs, machinelearning, macie, marketplace-entitlement, marketplacecommerceanalytics, mediaconvert, medialive, mediapackage, mediastore, mediastore-data, mediatailor, meteringmarketplace, mgh, mobile, mq, mturk, neptune, opsworks, opsworkscm, organizations, pi, pinpoint, pinpoint-email, polly, pricing, rds, redshift, rekognition, resource-groups, resourcegroupstaggingapi, route53, route53domains, s3, sagemaker, sagemaker-runtime, sdb, secretsmanager, serverlessrepo, servicecatalog, servicediscovery, ses, shield, signer, sms, snowball, sns, sqs, ssm, stepfunctions, storagegateway, sts, support, swf, transcribe, translate, waf, waf-regional, workdocs, workmail, workspaces, xray (略)

AWS Lambda Layersを利用して最新のAWS SDKを利用する方法は下記が参考になります。

AWS CloudFormationのAWS Lambda-backedカスタムリソースで最新のAWS SDKを利用する – Qiita https://cloudpack.media/48058

AMBリソースの更新・削除に対応する

AMBリソースはAWS Lambda-backedカスタムリソースで作成しているので、更新や削除もLambda-backedカスタムリソースで行う必要があります。

詳細は下記が参考になりますが、こちらもリソース作成の待受と同じく、作成とは別のカスタムリソースを定義する必要があります。

AWS CloudFormationのLambda-backedカスタムリソースでリソースの更新・削除をする方法 – Qiita https://cloudpack.media/48205

作成待ちと更新・削除を担うカスタムリソースは共通化できたので、最終的には下記のようなリソース定義となりました。

テンプレート抜粋

Resources: # ネットワーク作成用のカスタムリソース CreateBlockchainNetwork: Type: Custom::CustomResource # リソース作成待受とリソース情報取得・更新・削除用のカスタムリソース BlockchainNetwork: Type: Custom::CustomResource Properties: NetworkId: !GetAtt CreateBlockchainNetwork.NetworkId DependsOn: CreateBlockchainNetwork # リソース情報取得用のカスタムリソース BlockchainMember: Type: Custom::CustomResource Properties: NetworkId: !GetAtt BlockchainNetwork.Network.Id MemberId: !GetAtt CreateBlockchainNetwork.MemberId DependsOn: BlockchainNetwork # Peerノード作成用のカスタムリソース CreateBlockchainPeerNode: Type: Custom::CustomResource Properties: NetworkId: !GetAtt BlockchainNetwork.Network.Id MemberId: !GetAtt BlockchainMember.Member.Id DependsOn: BlockchainMember # リソース作成待受とリソース情報取得・更新・削除用のカスタムリソース BlockchainPeerNode: Type: Custom::CustomResource Properties: NetworkId: !GetAtt BlockchainNetwork.Network.Id MemberId: !GetAtt BlockchainMember.Member.Id NodeId: !GetAtt CreateBlockchainPeerNode.NodeId DependsOn: CreateBlockchainPeerNode

AWS lambda-backedカスタムリソースで返すリソース情報(JSON)に気をつける

AWS lambda-backedカスタムリソースのLambda関数ではcfnresponse.send(event, context, cfnresponse.SUCCESS, data)のようにして処理結果をCFnに返すことができます。

最後のパラメータdataはJSONとなり、CFnのFn::GetAttで参照可能です。

カスタムリソースの応答オブジェクト – AWS CloudFormation https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/crpg-ref-responses.html

応答で送信される、custom resource providerによって定義された名前と値のペア。ここで指定する値には、Fn::GetAtt を使用して、テンプレート内の名前でアクセスできます。

ただし、ネストされたJSONをdataに含めてもNetwork.Idのようにして参照できないため、AWS SDKでAMBのリソース情報を取得して得られるJSONをそのままでは返すことができませんでした。

AWS CloudFormationのLambda-BackedカスタムリソースでネストされたJSONを返しても参照できない – Qiita https://cloudpack.media/48318

なので、AWS SDKでAMBのリソース情報を取得して得られるJSONを加工する必要があります。 JSONのキーをNetwork.Idとすると、CFn側でもNetwork.Idと参照できるようにしています。

LambdaからCFnにネットワーク情報を返す例

import cfnresponse import boto3 import json from datetime import date, datetime def handler(event, context): client = boto3.client("managedblockchain") networkId = event['ResourceProperties']['NetworkId'] response = {} if event['RequestType'] == 'Create': network = client.get_network( NetworkId=networkId ) orderingServiceEndpoint = network['Network']['FrameworkAttributes']['Fabric']['OrderingServiceEndpoint'] vpcEndpointServiceName = network['Network']['VpcEndpointServiceName'] response = { "Network.Id": networkId, "Network.FrameworkAttributes.Fabric.OrderingServiceEndpoint": orderingServiceEndpoint, "Network.VpcEndpointServiceName": vpcEndpointServiceName } cfnresponse.send(event, context, cfnresponse.SUCCESS, response)

詳細は下記が参考になります。

AWS CloudFormationのLambda-BackedカスタムリソースでネストされてるっぽいJSONを返す方法 – Qiita https://cloudpack.media/48329

Hyperledger FabricのクライアントをEC2インスタンスで構築する

Hyperledger FabricのクライアントをEC2インスタンスで構築する定義は下記を参考にさせてもらいました。

awslabs/amazon-managed-blockchain-client-templates: AWS CloudFormation templates to provision Amazon EC2 instances and install and configure clients for use with blockchain frameworks in Amazon Managed Blockchain https://github.com/awslabs/amazon-managed-blockchain-client-templates

こちらはHyperledger FabricのクライアントとなるEC2インスタンスを作成するテンプレートでしたので、自前のテンプレートへ組み込み、AMBで必要となるリソースの作成後、インスタンスが作成されるようにしました。

ポイントとしては以下となります。

CFnのcfn-signalヘルパースクリプトで完了シグナルをCFnに返す

CFnでEC2インスタンスを作成すると、EC2インスタンスのステータスがRunningとなった時点でリソース作成完了となります。そのため、テンプレートのUserDataで定義しているコマンド実行でエラーとなっても正常完了扱いとなり不便だったので、CFnのcfn-signalヘルパースクリプトで完了シグナルをCFnに返すようにしました。

cfn-signalを利用するには、CreationPolicyを定義する必要があります。

EC2インスタンス定義_一部抜粋

BlockchainClient: Type: AWS::EC2::Instance UserData: Fn::Base64: Fn::Sub: - | #!/bin/bash (略) /opt/aws/bin/cfn-signal -e $? --stack ${AWS::StackName} --resource BlockchainClient --region ${AWS::Region} CreationPolicy: ResourceSignal: Timeout: PT20M

詳細は下記が参考になります。

cfn-signal – AWS CloudFormation https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/cfn-signal.html

Fn::Subで変数の取扱い

UserDataに指定する値はFn::Base64とFn::Subを用いて指定しています。 Fn::Subは文字列中に置き換えたい変数がある場合に利用する関数となっています。

Fn::Sub – AWS CloudFormation https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/intrinsic-function-reference-sub.html

テンプレートで、スタックを作成または更新するまで使用できない値を含むコマンドまたは出力を作成するために、この関数を使用できます。

UserDataを編集しているとシェル変数を利用したくなるケースがでてきますが、そのまま利用するとエラーとなるため注意が必要です。シェル変数の利用方法は下記が参考になりました。

CloudFormationの中のEC2のユーザーデータでシェル変数を使用する｜ DevelopersIO https://dev.classmethod.jp/cloud/aws/using-variables-in-ec2-user-data-in-cloudformation/

UserDataで変数を利用する例

UserData: Fn::Base64: Fn::Sub: - | #!/bin/bash # これはOK echo ${HOGE} HOGE2=hogehoge # これはエラーになる echo ${HOGE2} # こうするとOK echo ${!HOGE2} - { HOGE: hoge }

Hyperledger Fabricのcliでコマンドを実行するタイミングに気をつける

以下は、UserDataの後半部分の抜粋となります。ところどころでsleepコマンドを実行して待受けています。リソース作成を繰り返し試行錯誤した結果となりますが、主に下記の理由からとなります。

Peerノードが利用可能になるのを待ち受け

OrdererからPeerノードへのデータ送信待ち

UserData一部抜粋

(略) /usr/local/bin/docker-compose -f docker-compose-cli.yaml up -d sleep 5m # enroll fabric-ca-client enroll -u https://${ADMIN_USERNAME}:${ADMIN_PASSWORD}@${FABRIC_CA_ENDPOINT} --tls.certfiles /home/ec2-user/${FABRIC_CA_FILE} -M /home/ec2-user/admin-msp cp -r /home/ec2-user/admin-msp/signcerts /home/ec2-user/admin-msp/admincerts echo ' Organizations: (略) ' > /home/ec2-user/configtx.yaml docker exec cli configtxgen -outputCreateChannelTx /opt/home/mychannel.pb -profile OneOrgChannel -channelID mychannel --configPath /opt/home/ sleep 30s # Create Channel docker exec cli peer channel create -c mychannel -f /opt/home/mychannel.pb -o ${ORDERING_SERVICE_ENDPOINT} --cafile /opt/home/${FABRIC_CA_FILE} --tls sleep 30s docker exec cli peer channel join -b mychannel.block -o ${ORDERING_SERVICE_ENDPOINT} --cafile /opt/home/${FABRIC_CA_FILE} --tls sleep 30s # Install ChainCode docker exec cli peer chaincode install -n mycc -v v0 -p github.com/chaincode_example02/go docker exec cli peer chaincode instantiate -o ${ORDERING_SERVICE_ENDPOINT} -C mychannel -n mycc -v v0 -c '{"Args":["init","a","100","b","200"]}' --cafile /opt/home/${FABRIC_CA_FILE} --tls sleep 30s docker exec cli peer chaincode list --instantiated -o ${ORDERING_SERVICE_ENDPOINT} -C mychannel --cafile /opt/home/${FABRIC_CA_FILE} --tls /opt/aws/bin/cfn-signal -e $? --stack ${AWS::StackName} --resource BlockchainClient --region ${AWS::Region}

UserData の実行ログの確認方法

UserDataで指定したコマンドの実行ログが確認できないか調べてみたらしっかりと出力されていました。下記が参考になりました。

Linux インスタンスでの起動時のコマンドの実行 – Amazon Elastic Compute Cloud https://docs.aws.amazon.com/ja_jp/AWSEC2/latest/UserGuide/user-data.html

AWSのCloud initのログの場所 | しびら http://yamada.daiji.ro/blog/?p=191

EC2インスタンス内

$ cat /var/log/cloud-init-output.log (略) + docker exec cli peer channel join -b mychannel.block -o orderer.n-xxxxxxxxxxxxxxxxxxxxxxxxxx.managedblockchain.us-east-1.amazonaws.com:30001 --cafile /opt/home/managedblockchain-tls-chain.pem --tls 2019-07-01 09:17:03.663 UTC [channelCmd] InitCmdFactory -> INFO 001 Endorser and orderer connections initialized 2019-07-01 09:17:03.903 UTC [channelCmd] executeJoin -> INFO 002 Successfully submitted proposal to join channel + sleep 30s + docker exec cli peer chaincode install -n mycc -v v0 -p github.com/chaincode_example02/go 2019-07-01 09:17:34.076 UTC [chaincodeCmd] checkChaincodeCmdParams -> INFO 001 Using default escc 2019-07-01 09:17:34.076 UTC [chaincodeCmd] checkChaincodeCmdParams -> INFO 002 Using default vscc 2019-07-01 09:17:34.490 UTC [chaincodeCmd] install -> INFO 003 Installed remotely response:<status:200 payload:"OK" > + sleep 30s + docker exec cli peer chaincode instantiate -o orderer.n-xxxxxxxxxxxxxxxxxxxxxxxxxx.managedblockchain.us-east-1.amazonaws.com:30001 -C mychannel -n mycc -v v0 -c '{"Args":["init","a","100","b","200"]}' --cafile /opt/home/managedblockchain-tls-chain.pem --tls 2019-07-01 09:17:44.686 UTC [chaincodeCmd] checkChaincodeCmdParams -> INFO 001 Using default escc 2019-07-01 09:17:44.686 UTC [chaincodeCmd] checkChaincodeCmdParams -> INFO 002 Using default vscc + sleep 30s + docker exec cli peer chaincode list --instantiated -o orderer.n-xxxxxxxxxxxxxxxxxxxxxxxxxx.managedblockchain.us-east-1.amazonaws.com:30001 -C mychannel --cafile /opt/home/managedblockchain-tls-chain.pem --tls Get instantiated chaincodes on channel mychannel: Name: mycc, Version: v0, Path: github.com/chaincode_example02/go, Escc: escc, Vscc: vscc + /opt/aws/bin/cfn-signal -e 0 --stack amb-cfn-test --resource BlockchainClient --region us-east-1 Cloud-init v. 0.7.6 finished at Mon, 01 Jul 2019 09:19:16 +0000. Datasource DataSourceEc2. Up 692.70 seconds

まとめ

AMBのリソースをCFnで管理するのにAWS Lambda-backedカスタムリソースを利用することができましたが、そこそこハマるところがあり、テンプレート作成に時間がかかりました。 1度作成できたら応用を効かせることができそうですので、個人的には良いテンプレートができたなと思ってます^^

参考