こんにちは。株式会社船井総合研究所でデータ分析に携わっている平尾です。
今回は、Microsoft Fabricにおいてデータフレームが記載されているエクセルファイルを取り込み、テーブルを作成する方法に関してお伝えします。
また、そのデータフレームに不備があった際にTeamsでエラーメッセージを表示する方法も
お伝えします。
（pythonのライブラリであるpandasの基礎的な理解があることを前提にお話させていただきます。）

対象のエクセルファイル

今回は、「テーブル」と「補足」の２つのシートが存在する場合を例にします。
「テーブル」シート

「補足」シート

Notebookにおいてファイルを取り込む

まず、以下のようにエクセルファイルを格納します。

そして、Notebookにおいて以下のようなコードを記入します。

import pandas as pd
df1 = pd.read_excel("/lakehouse/default/Files/right/product1.xlsx", sheet_name = "テーブル" ,header=2, usecols="B:D") 
missing_rows = df1[df1['商品コード'].isnull()]  #'商品番号列において欠損値を含む行を抽出
min_row_index = missing_rows.index.min() #抽出された行のindexの最小値を取得
df2 = df1.iloc[:min_row_index] #抽出したい範囲のみ取得
df2['値段(円)'] = df2['値段(円)'].astype(int) # '値段(円)' の小数点を切り捨て
df2

すると、以下のようにファイルが取り込まれます。

「テーブル」シートの「商品コード」列において欠損値が発生するまでの行をデータフレームとして認識させています。そのため、「メモ」は除外されます。

データフレームを保存する

Notebookにおいて以下のようなコードを記入します。

df3 = pd.read_excel("/lakehouse/default/Files/right/product1.xlsx", sheet_name = "補足" ,header=1, usecols="B:C") 
from pyspark.sql import SparkSession

# SparkSessionを作成
spark = SparkSession.builder.appName("SaveDataFrameAsTable").getOrCreate()

# pandas DataFrameをSpark DataFrameに変換
spark_df = spark.createDataFrame(df2)

#保存先のテーブル名と形式を指定
table_name = f"`製品テーブル_{df3.iloc[0,0]}`" # テーブル名を指定
format = "parquet"  # 保存形式を指定 (parquet, csv, jsonなど)

# テーブルとして保存
spark_df.write.saveAsTable(table_name, format=format, mode="overwrite", orderBy="index")  # mode="overwrite" で既存のテーブルを上書き

すると、以下のように、「製品テーブル_平尾」というテーブルが作成されます。

ここでは、様々なデータソースへの接続、データの読み込み、変換、分析などを実行できる
「SparkSession」を使用しています。
また、エクセルファイルの「補足」シートの「記載者」を、テーブル名に使用しています。

取り込んだ際のエラーメッセージをTeamsに表示する。

上のように、「補足」シートにおいて欠損値があった際にエラーをTeamsに表示する方法を説明します。

まず、ここまでで書いたコードはこのようになっています。

import pandas as pd
df1 = pd.read_excel("/lakehouse/default/Files/right/product1.xlsx", sheet_name = "テーブル" ,header=2, usecols="B:D") 
missing_rows = df1[df1['商品コード'].isnull()]  #'商品番号列において欠損値を含む行を抽出
min_row_index = missing_rows.index.min() #抽出された行のindexの最小値を取得
df2 = df1.iloc[:min_row_index] #抽出したい範囲のみ取得
df2['値段(円)'] = df2['値段(円)'].astype(int) # '値段(円)' の小数点を切り捨て
df2


df3 = pd.read_excel("/lakehouse/default/Files/right/product1.xlsx", sheet_name = "補足" ,header=1, usecols="B:C") 
from pyspark.sql import SparkSession

# SparkSessionを作成
spark = SparkSession.builder.appName("SaveDataFrameAsTable").getOrCreate()

# pandas DataFrameをSpark DataFrameに変換
spark_df = spark.createDataFrame(df2)

#保存先のテーブル名と形式を指定
table_name = f"`製品テーブル_{df3.iloc[0,0]}`" # テーブル名を指定
format = "parquet"  # 保存形式を指定 (parquet, csv, jsonなど)

# テーブルとして保存
spark_df.write.saveAsTable(table_name, format=format, mode="overwrite", orderBy="index")  # mode="overwrite" で既存のテーブルを上書き

上のコードブロックにおける10行目のdf3 = .. を挟む形で、以下のコードを追加します。
(エラーを発生させるファイルを用いるために、ファイルパスを変えています。)

import requests
import json

def send_teams_message(message):
    webhook_url = "https:~"  # Teams の Webhook URL を設定 ※ここではURLを省略しています。
    headers = {"Content-Type": "application/json"}
    payload = {
        "text": f"Notebook でエラーが発生しました: {message}", # メッセージにメンションを追加,
        "markdown": True # markdown 形式を有効にする
    }
    response = requests.post(webhook_url, data=json.dumps(payload), headers=headers)
    if response.status_code != 200:
        print(f"Teams への通知に失敗しました: {response.text}")

df3 = pd.read_excel("/lakehouse/default/Files/wrong/product2.xlsx", sheet_name = "補足" ,header=1, usecols="B:C") 


missing_count = df3.isnull().sum().sum()

# 欠損値がある場合にエラーメッセージを表示
if missing_count > 0:
    # 欠損値がある列名を取得
    missing_cols = df3.columns[df3.isnull().any()].tolist()
    error_message = f"{', '.join(missing_cols)}が入力されていません"
    send_teams_message(error_message)
    raise ValueError(error_message)  # エラーを再発生させる (必要に応じて)

TeamsのwebhookURLを用いてエラーメッセージを送れるように設定しています。

webhookURLの作り方は、以下のものを参照してください。
https://learn.microsoft.com/ja-jp/microsoftteams/platform/webhooks-and-connectors/how-to/add-incoming-webhook?tabs=newteams%2Cdotnet

これで、もし「補足」シートに欠損値があった際には以下のメッセージが送られます。

コード全体

最終的なコード全体がこちらです。

import pandas as pd
df1 = pd.read_excel("/lakehouse/default/Files/wrong/product2.xlsx", sheet_name = "テーブル" ,header=2, usecols="B:D") 
missing_rows = df1[df1['商品コード'].isnull()]  #'商品番号列において欠損値を含む行を抽出
min_row_index = missing_rows.index.min() #抽出された行のindexの最小値を取得
df2 = df1.iloc[:min_row_index] #抽出したい範囲のみ取得
df2['値段(円)'] = df2['値段(円)'].astype(int) # '値段(円)' の小数点を切り捨て

import requests
import json

def send_teams_message(message):
    webhook_url = "https:~" # Teams の Webhook URL を設定 ※ここではURLを省略しています。
    headers = {"Content-Type": "application/json"}
    payload = {
        "text": f"Notebook でエラーが発生しました: {message}", # メッセージにメンションを追加,
        "markdown": True # markdown 形式を有効にする
    }
    response = requests.post(webhook_url, data=json.dumps(payload), headers=headers)
    if response.status_code != 200:
        print(f"Teams への通知に失敗しました: {response.text}")

df3 = pd.read_excel("/lakehouse/default/Files/wrong/product2.xlsx", sheet_name = "補足" ,header=1, usecols="B:C") 


missing_count = df3.isnull().sum().sum()

# 欠損値がある場合にエラーメッセージを表示
if missing_count > 0:
    # 欠損値がある列名を取得
    missing_cols = df3.columns[df3.isnull().any()].tolist()
    error_message = f"{', '.join(missing_cols)}が入力されていません"
    send_teams_message(error_message)
    raise ValueError(error_message)  # エラーを再発生させる (必要に応じて)



from pyspark.sql import SparkSession

# SparkSessionを作成
spark = SparkSession.builder.appName("SaveDataFrameAsTable").getOrCreate()

# pandas DataFrameをSpark DataFrameに変換
spark_df = spark.createDataFrame(df2)

#保存先のテーブル名と形式を指定
table_name = f"`製品テーブル_{df3.iloc[0,0]}`" # テーブル名を指定
format = "parquet"  # 保存形式を指定 (parquet, csv, jsonなど)

# テーブルとして保存
spark_df.write.saveAsTable(table_name, format=format, mode="overwrite", orderBy="index")  # mode="overwrite" で既存のテーブルを上書き

最後に

ここまで、Microsoft Fabricにおいてエクセルファイルを取り込み、データテーブルを保存する方法と、エラーメッセージを表示する方法に関してお伝えしました。
また、Teamsのチャネルのメンバーに対しメンションをつけてエラーメッセージを送る方法を現在検討中です。メンションを付ける方法が分かり次第また投稿します。
お読みいただきありがとうございました。