Microsoft Fabric のデータパイプラインでファイル最終変更日時などを取り込む

## はじめに

ファイルをバイナリコピーした際にコピーを実施した日時での更新日付になってしまうため、この情報を取り込む方法を記載します。

## 手順

やりたいことのイメージ

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/5c096c7d-c892-10ae-9e86-c8fd1dd2369f.png)

これら二つの課題を対策して、ファイル最終変更日時をテーブルに取り込みます。

本記事ではそれぞれの課題ごとにパイプラインを作っています。

### 準備

対象ファイルの状態

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/eba5ad27-d00c-a7a7-bdb7-15945ac3ca45.png)

9/6 午後 1 時と、9/24 午後 5 時 27 分 の2ファイルがアップされています。

### 1. Hive パーティション形式で変更日時を表現して保存するパイプライン

課題①となる最終変更日時をファルダ構造に持たせることで Landing で最終変更日時を保持します。

#### パイプラインの全体像

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/3e06ffd6-e101-53e1-0a00-9b28f56c667a.png)



#### Get files

パラメータ化しているコピーのソースディレクトリ (source_dir) のファイル一覧を取得します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/611a3a2f-3331-0304-bbd6-8853832ac61b.png)


出力例は以下のようにファイルの配列です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/61987580-b745-68fa-c0c6-2955c027ac00.png)


#### Filter

Get files では子フォルダまで取れる（存在している場合）のでファイルだけに絞り込みます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/de73e768-62cd-5476-138f-7f42022f7712.png)

出力例は以下のように絞られた配列です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/357c41ac-a5c0-2a95-5199-ee87a6641905.png)


#### ForEach

対象ディレクトリ内のファイルごとに処理を実行します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/2e387e15-c447-01b6-3652-d0df78da7c9c.png)

#### Get modify at

対象となったファイルの最終変更日時を取得します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/a715c145-fd1e-6471-1727-0ccdb1ce16de.png)

出力例は以下のように変更日時です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/e349a324-53ae-97d6-25ff-c5e36c05e83e.png)


#### Copy hive mod

対象ファイルを 出力先パラメータ(target_dir) の配下に、hive パーティションで変更日時を表現したフォルダ構造でコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/6e345851-81de-4529-62a7-c431c59fdeea.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/10a54d97-df45-d4c9-1615-ba356d72cf4f.png)

出力先の式

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/e5bacc4c-da79-168b-8764-f67b5ea15301.png)


```json:

@concat(
    pipeline().parameters.target_dir,'/'
    ,'audit__file_modify_at='
    ,activity('Get modify at').output.lastModified
)

```


#### 実行確認

出力先パラメータ(target_dir) を、  **landing/transactional/general_sample/sales/1.0/full/hive/audit__run_date=2024-09-24/audit__run_at_id=2024092423020936** という形で、ジョブの実行日付と実行日時は指定された状態で実行します。

実行結果は以下のようになり、<列名=値> の hive パーティション構造で出力されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/a51a16d4-5ffa-f1a6-6e27-be8fab1bdb1a.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/0b096bdd-57ed-0203-10bd-39e2119a69fa.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/889ad2f1-e7c5-ff34-e96c-ecf28980ea89.png)


:::note warn

audit__run_at_id , audit__run_date はパラメータで指定されており、このパイプラインで生成した文字列ではありません。

ジョブの実行日時が入力されることを想定した列です。

:::


### 2. Hive パーティションを検知して取り込むパイプライン

次は非常に簡単です。Hive パーティション形式でファイルが保存されているのでパーティション（フォルダ）の値を取り込み対象にすることができます。

#### ソース設定

lakehouse の ファイルをソースにして、ファイルパスを指定します。先ほどコピー先となったaudit__run_at_id=～のフォルダは以下のようになっているのでこれをソースにします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/87a99ba6-ae55-47b0-d12d-42f54ef11582.png)


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/f69eddd7-207e-23b8-fea2-eea56b27928c.png)

:::note info

運用時にはこのパスはパラメータ化してすると汎用化されます

:::

次にパーティション検出を有効にします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/5b206c61-c1b7-a289-be86-21625637ffed.png)

これにより、列名=値のフォルダ構造を利用した情報がレコードに付加されます。

プレビューを確認すると期待通りになっています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/5ebfe636-8e37-8d9e-b136-292ee8bf4492.png)

:::note info

audit__run_date もパーティション形式なので取得可能ですが、日々のバッチ処理で今回の処理日時外のファイルまで読み取られてしまうので外しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/c89a42b7-7cd1-1fa9-6792-90900a69c4b1.png)


:::


ついでに、ファイル名などの情報もレコードにしてしまいます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/9fef010f-b69f-91fd-8fa2-decef375ae62.png)

このあたりの仕様は [コピー中に列を追加する](https://learn.microsoft.com/ja-jp/azure/data-factory/copy-activity-overview#add-additional-columns-during-copy) を参照


#### あて先の指定

今回はテーブルを上書きするように構成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/39aaf812-c2b0-b2a2-1793-fa71285f7188.png)

#### マッピング

今回はマッピング指定なしで実施します。ソースで読み取られた列がすべてそのまま反映されます。


:::note info

型変換や、列名のマッピングを必要とする場合にはマッピングをします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/df5aff5b-f2b6-7c00-0a91-30db13b8180f.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/32ace8a0-3bbd-1828-ad0f-1a18360282f3.png)


:::


### 3. 結果確認

実行結果を確認します。

プレビュー

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/9575829f-7ebf-1f1c-235e-ca6ee2da759a.png)

ファイル名と最終変更日時の組み合わせ

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/8ee8e0cb-5023-8219-3c16-3a90c574f441.png)