自分が所属している会社のメンバーの教育用資料として、それなりの規模のデータを扱う時に前提として意識しておかなければいけないことをざっくりまとめたので、弊社特有の話は除外して公開用に整理してみました。 大規模データ処理、分散処理に慣れている人にとっては今更改めて言うことじゃないだろ、みたいな話ばかりだと思いますが、急激にデータスケールが増大してしまったりすると環境に開発者の意識が追い付かないこともあるかと思います。 そういったケースで参考にできるかもしれません。 弊社は基本的にAWSによって運用されているので、AWSを前提にした様なキーワードやサービス名が出てきます。後、句読点があったり無かったりしますが、ご容赦ください。 追記: 社内用の資料の編集なのでかなりハイコンテキストな内容だから誤解するかもしれませんが、これらはそもそもRDBの話ではありません。(関係無くは無いけど) 1000万オ
こんにちは、クラスメソッドの岡です。 今回Step Functionsを使って簡単なETL処理を試す機会があったので実際に作ったものを公開します。 サーバーレスでETL処理、といえばAWS Glueが浮かぶかと思いますが、今回はGlueは使わず、LambdaのPythonランタイムでPandasを使ってS3のデータとDynamoDBのデータを結合するような処理を行ってみたいと思います。 ちなみに私はデータ分析に関する知識はほぼ皆無ですが、PythonライブラリPandasを使う事で簡単にデータ処理を行えました。 シナリオ 今回はIoTデバイスから送られてくる時系列データがS3に出力されている前提として、そのファイルとDynamoDBにあるデバイスのマスタデータと結合して分析データとして別のS3バケットに出力する、といったシナリオを想定しています。 構成 サンプルコード 今回はServerl
最近、TreasureDataと自社のDBのデータを合わせた集計をバッチ処理することが多くpythonでいったんcsvに書き出してEmbulkでアップロードしていましたが、データソースが増えると面倒になっていくのでより簡単で汎用化した書き方を考えました。 結局はWF使ってやったらいい話なので需要があるか不明ですが、WF勉強するのも時間がかかるので慣れてる言語でてっとり早くETLを作りたい人向けの内容です。 前提条件 この記事が適している人 普段はpythonを書いているのでなるべくその範囲でETLしたい pandas-tdの動きが不安定すぎて使えない シェルでの実現方法を調べる時間もない 達成したい機能 SQLさえ作ればバッチを追加できるETLの箱を作る 入出力のテーブルが変わっても手間がかからない 記述した順番に処理が実行され、進捗が通知される 環境の準備 Embulkのインストール す
はじめに 昨年末にAirflowをさわってみてなかなか便利だと思いました。 【Airflow】最近よく聞くAirflowに入門！EC2で動かしてみた【CI/CD】 そこで次のステップとしてKubernetesとの連携に挑戦してみました。検索してみると「Airflow on Kubernetes」と呼ばれているようです。調べたかぎり、日本語の記事があまりなかったのでこの記事が役にたてばいいなと思います。 =====2020.1.7 追記===== 関連記事をリンクします。 【Airflow on Kubernetes】DockerイメージのbuildとPodのdeployの仕組みについて 【Airflow on Kubernetes】KubernetesPodOperatorの使い方 Airflow on Kubernetesとは ？ タスクスケジューリングやGUIでの実行管理が可能なAir
AirflowにはExecutorというものがあって、 使用するExecutorによって処理の方法が並列になるのか直列にするかを選ぶことができます。 今回はCeleryを使用したExecutorを選択しています。 上記のコマンドだけで以下にアクセスすることができます。 Airflow管理画面: localhost:8080 Flower: localhost:5555 実装 まずは個別の処理を書きます。 例として1日分のホテル情報をDBから取得してきてCSVに保存する処理を書いています。 import pandas import datetime from jinja2 import Template def make_hotel_table(**kwargs): # データ取得するsqlの実行 #実行日時の取得 execution_date = kwargs.get('execution
小さなファイルのETLにGlueを使うのがもったいなかったので、Pandasやpyarrowで実装しました。 Lambda Layerにpandasとpyarrowを追加 Layerに登録するパッケージを作成 パッケージをアップロード Lambdaのコード エラー対応 参考 Lambda Layerにpandasとpyarrowを追加 Layerに登録するパッケージを作成 今回利用するのはpandasとpyarrow、s3fsなのですが少し工夫が必要でした。 ３つを全てを一つのZIPに纏めるとLambda Layerの50MBの制限にかかってしまいます。 3つにZIPを分割するとLambdaにレイヤー追加する時の制限にかかってしまいます。 Layers consume more than the available size of 262144000 bytes 大きなnumpyなどを共有
Airflow - "Use airflow to author workflows as directed acyclic graphs (DAGs) of tasks. The airflow scheduler executes your tasks on an array of workers while following the specified dependencies. Rich command line utilities make performing complex surgeries on DAGs a snap. The rich user interface makes it easy to visualize pipelines running in production, monitor progress, and troubleshoot issues
1
リリース、障害情報などのサービスのお知らせ
最新の人気エントリーの配信
j次のブックマーク
k前のブックマーク
lあとで読む
eコメント一覧を開く
oページを開く
