AWS クラウド機械学習を使用したサーバーレスニュースデータパイプラインの構築

[[436699]]

[51CTO.com クイック翻訳]アナリストとして、私はニュースや業界の最新情報を追跡することに多くの時間を費やしています。私は産休中にこの疑問について考え、グリーンテクノロジーと再生可能エネルギーに関するニュースを追跡するシンプルなアプリを構築することにしました。 AWS Lambda と、EventBridge、SNS、DynamoDB、Sagemaker などの他の AWS サービスを使用すると、開始して数日でプロトタイプを構築するのは非常に簡単です。

このアプリケーションは、一連のサーバーレス Lambda 関数と、SageMaker エンドポイントとしてデプロイされたテキスト要約機械学習モデルによって実行されます。 AWS EventBridge ルールは、24 時間ごとに Lambda 関数をトリガーして、DynamoDB データベースからニュースフィードを取得します。

これらのニュースフィードは SNS トピックとして送信され、複数の Lambda がトリガーされてニュースフィードが分析され、ニュース URL が抽出されます。各サイトは、1 日に数件の記事のみで RSS ニュースフィードを更新します。そのため、特定のニュース発行のリソースを大量に消費する可能性のある大量のトラフィックは送信されません。

しかし、各ウェブサイトはそれぞれ異なるため、記事の全文を抽出することが大きな問題となります。幸いなことに、goose3 のようなライブラリは、機械学習の手法を適用してページの本文を抽出することでこの問題を解決します。著作権の問題により、記事の全文を保存することはできません。そのため、HuggingFace Text Summarizer モデルを適用して短い要約を生成しました。

ここでは、独自の機械学習ベースのニュース集約パイプラインを構築する方法について詳しく説明します。

1. 必要な権限を持つ IAM ロールを設定します。

このデータパイプラインはシンプルですが、多くの AWS リソースを接続します。関数に必要なすべてのリソースへのアクセスを許可するには、IAM ロールを設定する必要があります。このロールは、DynamoDB、Sagemaker、CloudWatch、SNS などの他のクラウドリソースを使用する権限を関数に付与します。セキュリティ上の理由から、IAM ロールに完全な AWS 管理アクセス権を与えるのではなく、必要なリソースの使用のみを許可するのが最適です。

2. RSS Dispatcher LambdaでDynamoDBからRSSニュースフィードを取得する

AWS Lambda を使用すると、ほぼ何でも実行できます。これは、短時間のタスクに最適な、非常に強力なサーバーレスコンピューティングサービスです。私にとっての主な利点は、AWS エコシステム内の他のサービスに簡単にアクセスできることです。

私はすべての RSS ニュースフィードを DynamoDB テーブルに保存しており、boto3 ライブラリを使用して Lambda からアクセスするのは非常に簡単です。データベースからすべてのニュースフィードを取得したら、それらを SNS メッセージとして送信し、ニュースフィード解析 Lambda をトリガーします。

 boto3をインポート
jsonをインポート
def lambda_handler(イベント、コンテキスト):
    ＃接続する  DynamoDBへ
    dynamodb = boto3.resource( 'dynamodb' ) 
     
    #テーブルを取得 
テーブル= dynamodb.Table ( 'rss_feeds ' ) 
     
    #テーブルからすべてのレコードを取得する 
    data = table .scan()[ 'アイテム' ]
    rss = [y[ 'rss' ]データのy ] 
     
    ＃接続する  SNSへ
    クライアント = boto3.client( 'sns' ) 
     
    # メッセージをキューに送信する
RSSのアイテムの場合:
        client.publish(TopicArn= "arn:aws:sns:eu-west-1:802099603194:rss_to-parse" 、メッセージ = item)

3. 必要なライブラリでレイヤーを作成する

AWS Lambda で特定のライブラリを使用するには、それらをレイヤーとしてインポートする必要があります。ライブラリをインポート用に準備するには、ライブラリを python.zip アーカイブに格納する必要があります。その後、ライブラリを AWS にアップロードして関数で使用できます。レイヤーを作成するには、Python フォルダーに cd し、pip install を実行して zip ファイルに圧縮し、アップロードの準備をします。

 pip インストール feedparser -t

しかし、goose3 ライブラリをレイヤーとして展開するのは困難です。簡単に調査したところ、LXML などの一部のライブラリは、Lambda のような環境 (Linux) でコンパイルする必要があることがわかりました。したがって、ライブラリを Windows でコンパイルしてから関数にインポートすると、エラーが発生します。この問題を解決するには、tarball を作成する前に Linux にライブラリをインストールする必要があります。

これを行うには 2 つの方法があります。まず、Docker を使用してシミュレートされた Lambda 環境にインストールします。私にとって最も簡単な方法は、AWS sam build コマンドを使用することです。関数がビルドされたら、ビルドフォルダーから必要なパッケージをコピーし、レイヤーとしてアップロードするだけです。

 sam ビルド--use-container

4. ニュースソースを解析するLambda関数を起動する

ニュース URL をトピックとして SNS に送信すると、複数の Lambda をトリガーして RSS ニュースフィードからニュース記事を取得できます。一部の RSS ニュースフィードは異なりますが、ニュースフィードパーサーライブラリを使用すると、さまざまな形式を使用できます。 URL はイベントオブジェクトの一部なので、キーで抽出する必要があります。

 boto3をインポート
フィードパーサーをインポートする
datetimeからdatetime をインポート
lambda_handler(イベント、コンテキスト): 
    
    ＃接続する  DynamoDBへ
    dynamodb = boto3.resource( 'dynamodb' )
    #テーブルを取得 
テーブル= dynamodb.Table ( 'ニュース' ) 
     
    # URLを取得する イベントから
    url = event[ '記録' ][0][ 'SNS' ][ 'メッセージ' ] 
    
    # RSSフィードを解析する
    フィード = feedparser.parse(url) 
     
フィード[ 'entries' ]内のアイテムの場合:
        結果 = {
 "news_url" : 項目[ 'link' ],
 "タイトル" : 項目[ 'タイトル' ],
 "created_at" : datetime.now().strftime( '%Y-%m-%d' ) # こうすることで、Dynamodb は日付を正しく処理できるようになります 
        } 
         
        # 結果をdynamodbに保存する
table .put_item(Item=result, ConditionExpression= 'attribute_not_exists(news_url)' ) # ストアのみ 固有のURL

5. Sagemakerでテキスト要約モデルを作成してデプロイする

Sagemaker は、AWS 上で機械学習モデルを簡単に作成、トレーニング、デプロイできるサービスです。 HuggingFace は AWS と提携して、ユーザーがモデルをクラウドに簡単にデプロイできるようにしました。

ここでは、Jupiter ノートブックに簡単なテキスト要約モデルを記述し、deploy() コマンドを使用してデプロイしました。

 sagemaker.huggingfaceからHuggingFaceModel をインポートします
sagemaker をインポートする
ロール = sagemaker.get_execution_role()
ハブ = {
 'HF_MODEL_ID' : 'facebook/bart-large-cnn' 、
 'HF_TASK' : '要約'  
 }
 # ハグフェイスモデルクラス
huggingface_model = ハギングフェイスモデル(
    トランスフォーマーバージョン = '4.6.1' 、
    pytorch_version = '1.7.1' 、
    py_version = 'py36' 、
    env=ハブ、
    役割=役割、
 ）
 # SageMaker Inferenceにモデルをデプロイする
予測子 = huggingface_model.deploy(
    initial_instance_count=1、#インスタンス数
    instance_type = 'ml.m5.xlarge' # ec2 インスタンスタイプ
）

デプロイされると、Sagemaker -> 推論 -> エンドポイント構成からエンドポイント情報を取得し、Lamdas で使用できるようになります。

6. 記事の全文と要約を取得し、結果をDynamoDBに保存する

著作権の関係で全文を保存していないため、すべての処理は Lambda で実行されます。 URL が Dynamo DB テーブルに格納されたら、テキスト処理 Lambda を起動します。これを実現するために、Lambda を起動するトリガーとして DynamoDB アイテム生成を作成しました。 Lambda が一度に 1 つの記事のみを処理するようにバッチサイズを作成しました。

 jsonをインポート
boto3をインポート
goose3からGooseをインポート
datetimeからdatetime をインポート
lambda_handler(イベント、コンテキスト):
    # DynamoDB レコード作成イベントからURL を取得する
    url = event[ 'レコード' ][ 0 ][ 'dynamodb' ][ 'キー' ][ 'news_url' ][ 'S' ] 
     
    #記事の全文を取得
    g = ガチョウ()
    記事 = g.extract(url=url) 
     
    body = article.cleaned_text # 記事のテキストをクリーンアップする
     
    published_date = article.publish_date #メタ記述から  
     
    # HuggingFaceテキスト要約モデルを使用して要約を作成する
    ENDPOINT_NAME = "your_model_endpoint"  
    ランタイム = boto3.client( 'runtime.sagemaker' ) 
     
    レスポンス =runtime.invoke_endpoint(EndpointName=ENDPOINT_NAME、ContentType= 'application/json' 、Body=json.dumps(data)) 
    
    #要約を抽出
    summary = json.loads(response[ 'Body' ] .read ()).decode()) です。 
     
    ＃接続する  DynamoDBへ
    dynamodb = boto3.resource( 'dynamodb' )
    #テーブルを取得 
テーブル= dynamodb.Table ( 'ニュース' ) 
     
    # dynamoDBに保存されているアイテムを更新する
更新=テーブル.update_item (
キー= { "news_url" : url }
        、
        条件式 = 'attribute_exists(news_url)' 、
        UpdateExpression = 'SET summary = :val1、published_date = :val2'  
        式属性値={
 ':val1' : 要約、
 ':val2' : 公開日
        }
    ）