Microsoft Fabric で Fabric Blog の RSS feed を AI 処理、可視化してみる

## はじめに

[Fabric Blog](https://blog.fabric.microsoft.com/en-US/blog) の RSS フィードを取得→サマリを翻訳＆サマリ内容からワークロードの分類をしてレポートにしたい

## 実装

メダリオンアーキテクチャにしてみます

### Bronze : 生データ保存

1. feed を取得した日を記録したいので、はじめにパラメータセルで run_date を設定
    ```python:pyspark
    
    run_date = '2024-12-13'
    
    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/55a430e8-3351-9f7c-8309-5d49454de2c0.png)

2. feedparser で RSS フィードを取得
    ```python:pyspark
    
    !pip install feedparser
    
    import feedparser
    
    # RSSフィードのURL
    rss_url = "https://blog.fabric.microsoft.com/en-us/blog/feed/"
    
    # フィードを解析
    feed = feedparser.parse(rss_url)
    feed
    
    ```
3. bronze テーブルに entry を加工せずに保存
    ```python:pyspark
    # bronze
    import pyspark.sql.functions as F
    import json 
    
    entries = []
    for entry in feed.entries:
        # エントリ全体をJSON形式で保存
        entries.append({
            "entry": json.dumps(entry)  # エントリ全体をJSON文字列として保持
        })
    df_bronze = spark.createDataFrame(entries)
    df_bronze = df_bronze.withColumn("run_date", F.to_date(F.lit(run_date)))
    
    # 動的パーティション上書きを設定して、読みこんだデータフレームに存在するパーティションのみ差し替え
    spark.conf.set('spark.sql.sources.partitionOverwriteMode','dynamic')
    
    df_bronze.write.partitionBy("run_date").mode("overwrite").saveAsTable("10_bronze_feed")
    
    display(df_bronze)
    
    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/3ce0ebf3-1db4-f088-4e08-0a1de4c0a394.png)
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/5023c2c9-9de5-8e88-a43b-c3505c2af553.png)

### Silver : エントリーを展開して UPSERT 型で蓄積

1. エントリーを展開、データ型も整備します。
    
    ```python:pyspark
    # silver
    from pyspark.sql.types import * 
    
    json_schema = StructType([
        StructField("id", StringType(), True),
        StructField("title", StringType(), True),
        StructField("link", StringType(), True),
        StructField("published", StringType(), True),
        StructField("summary", StringType(), True),
    ])
    
    # JSON列を解析して必要な情報を追加
    df_parsed = df.withColumn("entry_parsed", F.from_json(F.col("entry"), json_schema))
    
    
    # <p> と </p> の間の内容を抽出する
    df_cleaned = df_parsed.withColumn(
        "summary", 
        F.regexp_extract(F.col("entry_parsed.summary"), "<p>(.*?)</p>", 1)
    )
    
    # extracted_summary の中の &hellip; を実際の '…' に変換する
    df_cleaned = df_cleaned.withColumn(
        "summary", 
        F.regexp_replace(F.col("summary"), "&hellip;", "…")
    )
    
    spark.conf.set("spark.sql.legacy.timeParserPolicy", "LEGACY")
    
    df_cleaned = df_cleaned.withColumn(
        "published", 
        F.to_timestamp(F.col("entry_parsed.published"), "EEE, dd MMM yyyy HH:mm:ss 'Z'")
    )
    
    # 必要なカラムを抽出
    df_silver = df_cleaned.selectExpr(
        "entry_parsed.id",
        "entry_parsed.title",
        "entry_parsed.link",
        "published",
        "summary",
        "entry as raw_entry" 
    )    
            
    display(df_silver)
    ```

2. silver テーブルとして保存。初回のみ上書き。次回以降は MERGE で更新していきます。
    
    ```python:pyspark
    
      
        silver_table_name = "20_silver_feed"
        table_exists = spark.sql(f"SHOW TABLES LIKE '{silver_table_name}'").count() > 0
        
        if table_exists :
            print(f"MERGE")
        
            df_silver.createOrReplaceTempView("source_silver")
            
            spark.sql(f"""
            MERGE INTO {silver_table_name} AS target
            USING (
                SELECT * FROM source_silver 
            ) AS source
            ON target.id = source.id 
            WHEN MATCHED THEN
                UPDATE SET *
            WHEN NOT MATCHED THEN
                INSERT *    
            """)
        else : 
            print(f"CREATE TABLE")
            df_silver.write.mode("overwrite").saveAsTable(silver_table_name)

    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/3c0cdeb4-b21e-25e4-740c-c17b5cc2b191.png)

### Gold : Fabric の AI 機能で翻訳と分類をしてみる

[Microsoft Fabric の組み込みの AI モデルで自然言語に対する解析・分析をしてみる](https://qiita.com/ryoma-nagata/items/daeebd71e40312a42e22) を発展させてみます。たぶんそのうちこんな風にしなくてもAI分類とかできます。チャレンジ。

:::note info

F64 が必要になるので、 Azure Open AI などをデプロイしている場合には、[Bring Your Own Key を使用する SynapseML での Azure AI サービス](https://learn.microsoft.com/ja-jp/fabric/data-science/ai-services/ai-services-in-synapseml-bring-your-own-key) を参考に実装することになります。

:::


レポート用に スタースキーマまでやるのが筋ですが、今回は フィードの蓄積と AI 処理が動機だったので割愛しました。

1. silver データ取得 
    ```python:pyspark
    
    df_silver = spark.sql("SELECT * FROM LH_rss.20_silver_feed")
    display(df_silver)
    
    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/69c87d96-afbe-a9f2-2c7c-9256438ac76c.png)
1. AI処理は重めなので、Gold に存在しない ID を処理対象にします。
    ```python:pyspark
    
    gold_table_name = "30_gold_feed"
    table_exists = spark.sql(f"SHOW TABLES LIKE '{gold_table_name}'").count() > 0
    
    if table_exists:
        df_gold = spark.sql(f"SELECT * FROM {gold_table_name}")
        df_silver = df_silver.join(df_gold,(df_gold.id == df_silver.id) , "left_anti")
    
    if df_silver.count() != 0 :
        mssparkutils.notebook.exit("no update")
    
    ```
3. SynapseML の Translate で翻訳
    ```python:pyspark
    import synapse.ml.core
    from synapse.ml.cognitive.translate import *
    from synapse.ml.services.openai import OpenAIChatCompletion
    from pyspark.sql import Row
    from pyspark.sql.types import *
    from pyspark.sql.functions import col, struct, array ,lit, flatten
    
    # サマリーを翻訳
    
    translate = (Translate()
        .setTextCol("summary")
        .setToLanguage(["ja"])
        .setOutputCol("translation")
        .setConcurrency(5))
    
    df_translate = translate.transform(df_silver)\
            .withColumn("translation", flatten(col("translation.translations")))\
            .withColumn("translation", col("translation.text")[0])
    
    display(df_translate)
    
    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/6f91bdb3-b8f3-5aa7-cd2b-baa89b0b9047.png)
4. Chat Completion を応用して分類してみる。system_message にて、分類したいワークロードを記載してチャットでこれを聞いたというようなイメージ
    ```python:pyspark
    # 分類
    
    system_message= """Classify into one of the following categories. 
        Data Factory , 
        Data Warehouse , 
        Data Science, 
        Data Enginnering,
        Real-time Intelligence , 
        Power BI , 
        Other """
    
    # Create messages column
    df_chat = df_translate.withColumn(
        "messages",
        array(
            struct(lit("system").alias("role"), lit(system_message).alias("content"),lit("system").alias("name")),
            struct(lit("user").alias("role"), col("summary").alias("content"),lit("user").alias("name"))
        )
    )
    
    # Define the OpenAIChatCompletion transformation
    chat_completion = (
        OpenAIChatCompletion()
        .setDeploymentName("gpt-35-turbo-0125")  # deploymentName could be one of {gpt-35-turbo-0125 or gpt-4-32k}
        .setMessagesCol("messages")
        .setErrorCol("error")
        .setOutputCol("chat_completions")
    )
    
    # Apply the transformation to the summary DataFrame
    df_gold = chat_completion.transform(df_chat)\
        .select("id","title","link", "published","translation", col("chat_completions.choices.message.content")[0].alias("workload"))
    
    display(df_gold)
    
    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/ae495ff6-dc75-6287-b0bd-4e53ddb27fc5.png)
    分類できた感じがします。

5. 保存。
    gold テーブルに存在しないレコードで処理しているので Append で。
    
    ```python:pyspark
    df_gold.write.mode("append").saveAsTable(gold_table_name)
    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/da88d086-1817-46b5-60f9-aa4015cf7879.png)

### レポート

スタースキーマもなにもないテーブルですが、リンクで飛べるようにデータモデルの設定をしてから可視化します。

1. リンクで飛べるようにモデルを修正
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/73ebeed1-007a-68e0-a8a5-a2c79973414f.png)
2. レポート完成
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/340151a6-a420-26b4-66a7-c15c166c6940.png)

### ジョブ化


1. ジョブ化する場合には、 ノートブックセッションでの !pip install は使えないので、環境を作成してライブラリを仕込みます。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/110ee2b9-5af2-a633-3616-e85bb49ff14f.png)
1. ノートブックからは install を削除して、環境を設定
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/a26fee2d-a760-b2a7-7ffa-a7419c1ad9d9.png)
1. パイプラインでノートブックを呼び出すようにします。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/78e0a9d7-14f4-5020-8f52-ba45b4b1e2be.png)
1. run_date を設定できるようにします。
    ```
    
    formatDateTime(convertFromUtc(utcnow(),'Tokyo Standard Time'),'yyyy-MM-dd')
    
    ```

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/4252f7a9-bef3-bcbd-a4ec-adeaec82493d.png)
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/821578c1-33ef-c16b-ae89-3021472d57cc.png)
1. スケジュール設定をしたら完成
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/7a251adf-ff70-5107-8f43-9c2dfda4f921.png)