packages:
  - package: dbt-labs/dbt_external_tables
    version: 0.11.1

[
  {
    "id": 123,
    "user": {
      "name": "John",
      "email": "john@example.com"
    },
    "tags": [
      "tech",
      "cloud"
    ],
    "update_at": "2025-01-01 00:00:00"
  },
  {
    "id": 124,
    "user": null,
    "tags": [
      "data",
      "analytics"
    ],
    "update_at": "2025-01-02 12:00:00"
  }
]

import json
import pandas as pd

# JSON ファイルを読み込む
with open("sampile.json", "r", encoding="utf-8") as infile:
    data = json.load(infile)

# フラット化（ネストを解除）
df = pd.json_normalize(data)

# Parquet ファイルに変換（pyarrowを使用）
df.to_parquet("sampile.parquet", engine="pyarrow", index=False)

{"id":123,"tags":["tech","cloud"],"update_at":"2025-01-01 00:00:00","user.name":"John","user.email":"john@example.com","user":null}
{"id":124,"tags":["data","analytics"],"update_at":"2025-01-02 12:00:00","user.name":null,"user.email":null,"user":null}

import json
import pandas as pd

# JSON ファイルを読み込む
with open("sampile.json", "r", encoding="utf-8") as infile:
    data = json.load(infile)

df = pd.json_normalize(data)
df.columns = df.columns.str.replace('.', '__')

# Parquet ファイルに変換（pyarrowを使用）
df.to_parquet("sampile.parquet", engine="pyarrow", index=False)

{"id":123,"tags":["tech","cloud"],"update_at":"2025-01-01 00:00:00","user__name":"John","user__email":"john@example.com","user":null}
{"id":124,"tags":["data","analytics"],"update_at":"2025-01-02 12:00:00","user__name":null,"user__email":null,"user":null}

import json
import pandas as pd

# JSON ファイルを読み込む
with open("sampile.json", "r", encoding="utf-8") as infile:
    data = json.load(infile)

def convert_nested_objects_to_string(record):
    for key, value in record.items():
        if not isinstance(value, (int, float, str, bool, type(None))): # 単一の値かどうかを判定
            record[key] = json.dumps(value, ensure_ascii=False)  # JSON 文字列に変換
    return record

# すべてのレコードに対して変換を適用
data = [convert_nested_objects_to_string(record) for record in data]

# フラット化（ネストを解除）
df = pd.json_normalize(data)

# Parquet ファイルに変換（pyarrowを使用）
df.to_parquet("sampile.parquet", engine="pyarrow", index=False)

{"id":123,"user":"{\"name\": \"John\", \"email\": \"john@example.com\"}","tags":"[\"tech\", \"cloud\"]","update_at":"2025-01-01 00:00:00"}
{"id":124,"user":null,"tags":"[\"data\", \"analytics\"]","update_at":"2025-01-02 12:00:00"}

import json
import pandas as pd
from datetime import datetime

# JSON ファイルを読み込む
with open("sampile.json", "r", encoding="utf-8") as infile:
    data = json.load(infile)

def convert_nested_objects_to_string(record):
    for key, value in record.items():
        if not isinstance(value, (int, float, str, bool, type(None))): # 単一の値かどうかを判定
            record[key] = json.dumps(value, ensure_ascii=False)  # JSON 文字列に変換
    return record

# すべてのレコードに対して変換を適用
data = [convert_nested_objects_to_string(record) for record in data]

# フラット化（ネストを解除）
df = pd.json_normalize(data)
df["imported_at"] = datetime.utcnow()  # UTCで統一
print(df)

# Parquet ファイルに変換（pyarrowを使用）
df.to_parquet("sampile.parquet", engine="pyarrow", index=False)

version: 2
sources:
  - name: work
    database: ~
    schema: work
    tags:
      - porters
    loader: gcloud storage


    tables:
      - name: event
        external:
          location: 'gs://~output.parquet'
          options:
            format: parquet
        freshness:
          error_after: {count: 3, period: hour} 
        loaded_at_field: imported_at

dbt run-operation stage_external_sources --args "select: work" --vars "ext_full_refresh: true"
dbt source freshness --select source:work

    - name: field
       data_tests:
         - unique:
             column_name: id
         - not_null:
             column_name: id

      - name: field
        columns:
          - name: id
            data_tests:
              - unique
              - not_null

data_tests:
  +store_failures: true  # all tests
  
  <package_name>:
    +store_failures: false # tests in <package_name>

dbt run-operation stage_external_sources --args "select: 〜〜〜" --vars "ext_full_refresh: true"

dbt source freshness --select source:〜〜〜

dbt build --select tag:〜〜〜,tag:snapshot --store-failures

{% macro my_macro() %}
    {% set old_relation = source("sample", "field") %}
    {% set dbt_relation = ref("sample_snapshot__field") %}

    {% set excluded_columns = [
        "dbt_scd_id",
        "dbt_updated_at",
        "dbt_valid_from",
        "dbt_valid_to",
        "gcs_imported_at",
    ] %}

    {% set comparison_query %}
    {{ audit_helper.compare_relation_columns(
        a_relation=old_relation,
        b_relation=dbt_relation
    ) }}
    {% endset %}

    {% set sql_query %}
with
    check_list as (
        select *
        from ({{ comparison_query }})
        where not column_name in ('{{ excluded_columns | join("', '") }}')
    )

select countif(in_both = false) = 0 as validation_result
from
    check_list
    {% endset %}

    {% set results = run_query(sql_query) %}

    {% if execute %}
        {% set validation_result = results.columns[0].values()[0] %}
        {% if validation_result %}
            {% do log(
                "✅ データ型チェック成功：すべてのカラム整合性が確認されました",
                info=True,
            ) %}
        {% else %}
            {% do log(
                "❌ データ型不一致：check_list CTEで不一致カラムを確認してください",
                info=True,
            ) %}
        {% endif %}
    {% endif %}
{% endmacro %}

$ dbt run-operation my_macro
 ✅ データ型チェック成功：すべてのカラム整合性が確認されました

version: 2
sources:
- name: cricinfo
  database: "{{ env_var('ENV_CODE') | trim | upper }}_{{ env_var('PROJ_CODE') | trim | upper }}_RAW_DB"
  schema: "CRICSHEET"
  loader: S3
  tables:
    - name: all_player_details
      ext_full_refresh: true
      external:
        location: "@{{ env_var('ENV_CODE') | trim | upper }}_{{ env_var('PROJ_CODE') | trim | upper }}_RAW_DB.CRICSHEET.TST_ENTECHLOG_CRICSHEET_S3_STG/cricinfo/"  # required: S3 file path, GCS file path, Snowflake stage, Synapse data source 
        file_format: >
          (TYPE = JSON)
        partitions:
          - name: file_name
            data_type: varchar
            expression: metadata$filename
            
        columns:
          - name: player_data
            data_type: varchar

vars:
  # dbt-external-tables
  ext_full_refresh: true

{% test data_freshness(
    model, column_name, error_after_count, error_after_period, filter
) %}

    with
        freshness_check as (
            select max({{ column_name }}) as max_loaded_at
            from {{ model }}
            {% if filter %} where {{ filter }} {% endif %}
        )

    select
        max_loaded_at,
        {{ dbt.current_timestamp() }} as executed_at,
        {{ dbt.datediff("max_loaded_at", dbt.current_timestamp(), error_after_period) }}
        as diff_count
    from freshness_check
    where
        {{ dbt.datediff("max_loaded_at", dbt.current_timestamp(), error_after_period) }}
        >= {{ error_after_count }}  -- エラーになる閾値

{% endtest %}

version: 2

models:
  - name: orders
    columns:
      - name: last_updated_at
        tests:
          - data_freshness:
              error_after_count: 3   # エラーになる閾値
              error_after_period: "hour"  # 時間単位（hour/day/minuteなど）

# 8時間以上経過したらエラー
error_after_count: 8
error_after_period: "hour"

# 3日以上経過したらエラー
error_after_count: 3
error_after_period: "day"

# 15分以上経過したらエラー
error_after_count: 15
error_after_period: "minute"

version: 2

sources:
  - name: ~~
    database: ~~
    schema: ~~
    loader: gcloud storage


    tables:
      - name: ~~
        external:
          location: 'gs://~~~~.parquet'
          options:
            format: parquet
        freshness:
          warn_after: {count: 5, period: hour}
          error_after: {count: 15, period: hour} 
        loaded_at_field: gcs_imported_at
        data_tests:
          - unique:
              column_name: id
          - not_null:
              column_name: id
          - data_freshness:
              column_name: gcs_imported_at
              error_after_count: 15
              error_after_period: hour

{% set old_relation = source("ソーステーブル") %}
{% set dbt_relation = ref("スナップショットテーブル") %}

{% set excluded_columns = [
    "dbt_scd_id",
    "dbt_updated_at",
    "dbt_valid_from",
    "dbt_valid_to",
    "gcs_imported_at",
] %}


{% set comparison_query %}
    {{ audit_helper.compare_relation_columns(
        a_relation=old_relation,
        b_relation=dbt_relation
    ) }}
{% endset %}

{% set check_col = "null" %}

with
    check_list as (
        select *
        from ({{ comparison_query }})
        where not column_name in ('{{ excluded_columns | join("', '") }}')
    ),

    has_match_list as (  -- マッチしていない項目確認
        select *
        from check_list
        where not (has_ordinal_position_match and has_data_type_match)
    )

    {% if check_col %}
        ,
        col_check_list as (  -- マッチしていないカラムの内容
            select {{ check_col }}
            from {{ old_relation }}
            where {{ check_col }} is not null
        )
    {% endif %}

select *
from has_match_list

alter table `ucchi_test_snapshot`
rename column sex to sex_old;

alter table `ucchi_test_snapshot`
add column sex string;

update `ucchi_test_snapshot`
set sex = cast(sex_old as string)
where true;

alter table `ucchi_test_snapshot`
drop column sex_old;

import json
import datetime
import pandas as pd
from typing import Dict, Any, List

from util.log import log


def convert_json_to_parquet(data, output_parquet_path) -> None:
    """
    JSONファイルを読み込み、Parquet形式に変換して保存する関数。
    """

    log("INFO", f"Start converting JSON to Parquet: {output_parquet_path}")
    # キーのピリオドをアンダースコアに変換する
    data = replace_keys(data)

    # すべてのレコードに対して変換を適用
    data = [convert_nested_objects_to_string(record) for record in data]

    # フラット化（ネストを解除）
    df = pd.json_normalize(data)

    # 全カラムを nullable string に変換、nullを保持したまま文字列型にできる
    for col in df.columns:
        df[col] = df[col].astype("string") 

    # 現在のUTC日時を追加
    df["gcs_imported_at"] = datetime.datetime.now(datetime.timezone.utc)

    # Parquet ファイルに変換（pyarrowを使用）
    df.to_parquet(output_parquet_path, engine="pyarrow", index=False)

    log("INFO", f"Parquet file saved: {output_parquet_path}")


def convert_nested_objects_to_string(record):
    """
    ネストされたオブジェクトをJSON文字列に変換する関数

    Args:
        record (dict): 変換対象のレコード

    Returns:
        dict: ネストされたオブジェクトをJSON文字列に変換したレコード
    """
    for key, value in record.items():
        if not isinstance(value, (int, float, str, bool, type(None))):  # 単一の値かどうかを判定
            record[key] = json.dumps(value, ensure_ascii=False)  # JSON 文字列に変換
    return record


def replace_keys(obj: Any) -> Any:
    """オブジェクト内のすべてのキーのピリオドをアンダースコアに置換します。

    Args:
        obj (Any): 処理対象のオブジェクト。辞書、リスト、またはその他の型が可能です。

    Returns:
        Any: キーのピリオドがアンダースコアに置換された新しいオブジェクト。
            入力が辞書やリストでない場合は、入力をそのまま返します。

    Examples:
        >>> replace_keys({"a.b": 1, "c": {"d.e": 2}})
        {'a_b': 1, 'c': {'d_e': 2}}
    """
    if isinstance(obj, dict):
        new_obj = {}
        for key, value in obj.items():
            new_key = key.replace(".", "_").lower()
            new_obj[new_key] = replace_keys(value)
        return new_obj
    elif isinstance(obj, list):
        return [replace_keys(item) for item in obj]
    else:
        return obj

$ dbt run-operation stage_external_sources

＃ 既存のものがあればCREATE EXTERNAL TABLEされない
1 of 1 START external source ~.~
1 of 1 SKIP

$ dbt run-operation stage_external_sources --vars "ext_full_refresh: true"

＃ CREATE EXTERNAL TABLEされる
1 of 1 START external source porters.option
1 of 1 (1) create or replace external table `〜`.`〜`.`〜`              ... 
1 of 1 (1) None (0 processed)

vars:
	 # dbt-external-tables
  ext_full_refresh: true

    columns:
      - name: id
        data_type: STRING

create or replace external table `~`.`~`.`~`(
    id STRING
    )
    options (
        uris = ['gs://~_*.parquet']
        , format = 'parquet'
    )

dbt-osmosis yaml refactor

　meta:
      dbt-osmosis-skip-add-columns: true

version: 2

sources:
  - name: ~~
    database: ~~
    schema: ~~
    loader: gcloud storage
　meta:
      dbt-osmosis-skip-add-columns: true


    tables:
      - name: ~~
        external:
          location: 'gs://~~~~.parquet'
          options:
            format: parquet
        freshness:
          warn_after: {count: 5, period: hour}
          error_after: {count: 15, period: hour} 
        loaded_at_field: gcs_imported_at
        data_tests:
          - unique:
              column_name: id
          - not_null:
              column_name: id
          - data_freshness:
              column_name: gcs_imported_at
              error_after_count: 15
              error_after_period: hour
　　　columns: []