はじめに
「Power BI セマンティックモデルは分析エンジンなので、テーブル出力としては最適化されていない」とはいうものの、根強い csv 出力へのニーズがあります。
ページ分割レポートなどを使用してある程度の規模のテーブル出力には対応できますが、今回は 他システムへの連携を意識してコードで ファイル出力する案として、Semantic link を使用してみます。
準備
10,000,000 件のデータをインポートしたセマンティックモデルを使用します。
spark sql による抽出
参考：https://learn.microsoft.com/en-us/fabric/data-science/read-write-power-bi-spark
以下のようなクエリでセマンティックモデルにアクセスできます。
select * from pbi.`セマンティックモデル名`.<エンドポイント種類>.`テーブル名`
エンドポイント種類は以下が記入可能です。
- xmla : 省略した場合のデフォルト。 XMLA エンドポイントを使用
- rest : Power BI REST API に送信
- onelake : OneLake 統合または Direct Lake モードでのみ利用可能。直接 Delta Parquet にクエリする※確認中
spark sql での取得
まずは取得してみます。
ワークスぺース上のセマンティックモデルは pbi という名前空間に格納されています。
df = spark.sql("SHOW TABLES FROM pbi")
display(df)
対象のテーブルがわかったので、クエリしてみます。（ちょっと負荷のかかりそうな連番付与をしています）
df_spark=spark.sql("select *,row_number() OVER (order by S_SUPPKEY) as row_num from pbi.tpch_sample.TPCH_SF1000_SUPPLIER")
df_spark.count()
結果を保存します。
df_spark.write.format("csv").save("Files/sparksql_tpchsample_csv/data.csv")
spark sql の使用するセマンティックモデル用のエンドポイントについて
画像では抽出できてしまってますが、rest エンドポイントを使用した場合には制限事項が適用されます。
このような Power BI REST API に対する制限は通常、 XMLA エンドポイントを使用して回避します。
https://learn.microsoft.com/ja-jp/rest/api/power-bi/datasets/execute-queries#limitations
spark to csv について
spark は内部の処理でデータを分割してしまうため、使用される処理の状況により csv が分割されることがあります。
明示的に1ファイルにしたい場合や、逆に数を指定して分割したい場合、は以下のようにします。
※分割した場合、ファイル名は指定できません。
df_spark.repartition(4).write.format("csv").save("Files/part_sparksql_tpchsample_csv")
# または以下など
df_spark.coalesce(1).write.format("csv").save("Files/part_sparksql_tpchsample_csv/data.csv")
sempy による抽出
evaluate_dax と、evaluate_measureを確認します。
参考：
- https://learn.microsoft.com/ja-jp/fabric/data-science/read-write-power-bi-python
- https://learn.microsoft.com/ja-jp/fabric/data-science/tutorial-power-bi-measures
なお、read_table の場合は、エンドポイントの指定が可能です。
evaluate_dax
dax の結果を使用する方法を試します。
Power BI Desktop から DAX は簡単に確認 できるため、Power BI ユーザーはこの方法が使いやすいと思います。
この部分の DAX を使用してみます。ただし、レポートで生成される DAX には TOPN が使われているので、そこ以外を流用します
import sempy.fabric as fabric
dax= """
DEFINE
VAR __DS0Core =
SUMMARIZECOLUMNS(
'TPCH_SF1000_SUPPLIER'[S_NAME],
"sample_measure", 'TPCH_SF1000_SUPPLIER'[sample_measure]
)
EVALUATE
__DS0Core
ORDER BY
[sample_measure] DESC, 'TPCH_SF1000_SUPPLIER'[S_NAME]
"""
df_dax_result = fabric.evaluate_dax(
"tpch_sample",
dax
)
df_dax_result.count()
内容も見てみます。
display(df_dax_result)
出力します。
import os
folder_path = '/lakehouse/default/Files/evaluatedax_tpchsample_csv'
file_path = os.path.join(folder_path, 'data.csv')
if not os.path.exists(folder_path):
os.makedirs(folder_path)
df_dax_result.to_csv(file_path)
sempy のデータフレームは pandas と同じように動作するため、ファイルシステムのAPI パス を出力先にします。
evaluate_measure
XMLA エンドポイントを利用せずにバックエンド制限もない特殊な API を使用している模様です。
明示的に XMLA エンドポイントを使わせるには use_xmla=True を使用します。
メジャーの評価結果を使用する方法を試します。
先ほどの DAX と同じような結果となる内容で作成します。
use_xmla を使わないと非常にレスポンスが重かったためこれを使用しています。
import sempy.fabric as fabric
df_measure_result = fabric.evaluate_measure(
"tpch_sample",
"sample_measure",
["'TPCH_SF1000_SUPPLIER'[S_NAME]"],
use_xmla=True
)
df_measure_result.count()
結果も見てみます。
display(df_measure_result)
最後に出力
出力されたファイルの使用について
レイクハウスに出力されたファイルは Azure Data Lake Storage Gen2 と同じプロトコルでアクセスできるので、azcopy などを使用して転送してもよいですし、Onelake File Explorer （Preview） を使用して、ローカルファイルとしてアクセスすることも可能です。
