使用Power BI Python实现令人惊叹的数据可视化 – techsyncer

您的管理层是否想要了解公司的财务和生产力的方方面面，但却不愿意花一分钱购买顶级的IT管理工具？不要再为了库存、计费和票务系统而使用不同的工具。您只需要一个集中的系统。为什么不考虑使用Power BI Python呢？

Power BI可以将繁琐而耗时的任务转变为自动化的过程。在本教程中，您将学习如何以您无法想象的方式切片和组合您的数据。

快来吧，节省自己在复杂报告中费眼睛的压力吧！

先决条件

本教程将进行实际演示。如果您想跟着做，请确保您具备以下条件：

Power BI 订阅 – 免费试用即可。

A Windows Server – This tutorial uses a Windows Server 2022.

已安装在您的Windows Server 上的Power BI Desktop – 本教程使用Power BI Desktop v2.105.664.0。

已安装的MySQL Server – 本教程使用MySQL Server v8.0.29。

已在计划使用桌面版本的外部设备上安装本地数据网关。

Visual Studio Code（简称 VS Code）- 本教程使用 VS Code v17.2

已安装 Python v3.6 或更新版本 – 本教程使用 Python v3.10.5。

已安装 DBeaver – 本教程使用 DBeaver v22.0.2。

构建 MySQL 数据库

Power BI 可以美观地可视化数据，但在进行数据可视化之前，您需要获取和存储数据。存储数据的最佳方式之一是使用数据库。MySQL 是一个免费且功能强大的数据库工具。

1. 以管理员身份打开命令提示符，运行下面的 mysql 命令，并在提示时输入 root 用户名 (-u) 和密码 (-p)。

默认情况下，只有 root 用户有权限对数据库进行更改。

mysql -u root -p

2. 接下来，运行以下查询以创建一个新的数据库用户（CREATE USER），并设置密码（IDENTIFIED BY）。您可以为用户命名，但本教程选择的名称是 ata_levi。

CREATE USER 'ata_levi'@'localhost' IDENTIFIED BY 'Password';

3. 创建用户后，运行以下查询以授予新用户权限（ALL PRIVILEGES），例如在服务器上创建数据库。

GRANT ALL PRIVILEGES ON . TO 'ata_levi'@'localhost' WITH GRANT OPTION;

Granting privileges to the newly-created user

4. 现在，运行以下 \q 命令以注销 MySQL。

5. 运行以下 mysql 命令以使用新创建的数据库用户（ata_levi）登录。

mysql -u ata_levi -p

6. 最后，运行以下查询以创建名为 ata_database 的新数据库。当然，您可以为数据库取不同的名称。

CREATE DATABASE ata_database;

使用 DBeaver 管理 MySQL 数据库

在管理数据库时，通常需要具备 SQL 知识。但是使用 DBeaver，您可以通过几次点击来管理您的数据库，并且 DBeaver 将为您处理 SQL 语句。

1. 从桌面或开始菜单打开 DBeaver。

2. 当 DBeaver 打开时，单击新数据库连接下拉菜单，并选择 MySQL 以连接到您的 MySQL 服务器。

3. 使用以下方式登录到本地 MySQL 服务器：

将 服务器主机 保留为 localhost，并将端口设置为 3306，因为您正在连接到本地服务器。

提供来自“构建 MySQL 数据库”部分第二步的 ata_levi 用户的凭据（用户名和密码），然后单击“完成”以登录到 MySQL。

4. 现在，在数据库导航器（左侧面板）下扩展您的数据库（ata_database） → 右键单击 Tables，然后选择创建新表来初始化创建新表。

Creating a new database table with DBeaver GUI.

5. 给您的新表取个名字，但本教程的选择是 ata_Table，如下所示。

确保表名与“获取和使用 API 数据”部分第七步中将在 to_sql（“表名”）方法中指定的表名相匹配。

6. 接下来，扩展新表（ata_table） → 右键单击 Columns → 创建新列以创建新列。

7. 提供列名，如下所示，并选中 Not Null 复选框，然后点击确定以创建新列。

理想情况下，您可能想添加一个名为“id”的列。为什么呢？因为大多数 API 都会有一个 id，而 Python 的 pandas 数据框会自动填充其他列。

8. 点击保存（右下角）或按 Ctrl+S 保存更改，一旦您已验证新创建的列（id）。如下所示。

9. 最后，点击 Persist 以将您对数据库所做的更改持久保存。

获取和使用 API 数据

既然您已经创建了用于存储数据的数据库，您需要从相应的 API 提供者那里获取数据，并使用 Python 将其推送到您的数据库。您将源化数据以在 Power BI 上进行可视化。

要连接到 API 提供者，您将需要三个关键信息：授权方法、API 基本 URL 和 API 端点。如果对此信息或如何获取此信息有疑问，请访问您的 API 提供者的文档站点。

以下是来自Syncro的文档页面。

1. 打开VS Code，创建一个Python文件，并根据文件预期的API数据命名文件。该文件将负责从API获取数据并将数据推送到您的数据库（数据库连接）。

有多个Python库可用于帮助进行数据库连接，但在本教程中，您将使用SQLAalchemy。

在VS Code的终端上运行下面的pip命令，以在您的环境中安装SQLAalchemy。

pip install SQLAlchemy

2. 接下来，创建一个名为connection.py的文件，填写下面的代码，替换相应的值，并保存文件。

在开始编写与数据库通信的脚本之前，必须在数据库接受任何命令之前建立与数据库的连接。

但是，不必为您编写的每个脚本重新编写数据库连接字符串，下面的代码致力于使此连接能够被其他脚本调用/引用。

# 导入模块
# 从SQLAlchemy导入连接引擎
from sqlalchemy import create_engine
import sqlalchemy as db
from sqlalchemy.exc import SQLAlchemyError

# 使用try/except逻辑创建MySQL数据库连接
# 提供以下MySQL服务器凭据：
# db: 数据库名称
# user: 用户名
# pw: 密码
try:
    engine = create_engine("mysql+pymysql://{user}:{pw}@localhost/{db}"
    .format(
        db="ata_database",
        user="ata_levi",
        pw="password"
    ))
    print("Connected")
except SQLAlchemyError as e:
    error = str(e.dict['orig'])
print(error)

3. 打开 Visual Studio 的终端（Ctrl+Shift+`），并运行以下命令安装 pandas 和 requests。

pip install pandas
pip install requests

4. 创建另一个名为 invoices.py 的 Python 文件（或以其他名称命名），并将下面的代码填充到文件中。

您将在随后的每个步骤中向 invoices.py 文件添加代码片段，但您可以在 ATA 的 GitHub 上查看完整的代码。

invoices.py 脚本将从下一节中描述的主要脚本中运行，该主要脚本会拉取您的第一个 API 数据。

以下代码执行以下操作：

消费来自您的 API 的数据并将其写入您的数据库。

用您的 API 提供者凭据替换授权方法、密钥、基本 URL 和 API 端点。

# 导入必要的模块
import pymysql
from pandas.core.frame import DataFrame
import requests
import pandas as pd
from requests.api import get
from sqlalchemy.dialects.mysql import LONGTEXT

import os
import schedule
import time
import asyncio

# 从 connection.py 中导入连接字符串
from connection import engine

5. 将下面的代码片段添加到 invoices.py 文件中以定义标题，例如：

您期望从您的 API 接收的数据格式类型。

基本 URL 和端点应该与授权方法和相应的密钥一起。

请务必将下面的值更改为您自己的值。

# 定义请求参数。
headers = {
'accept':'application/json',
'authorization':'Bearer *Your-bearer-code if the authentication is Bearer*'
}
baseurl = 'https://*your base url*'
endpoint  = 'your endpoint'

6. 接下来，将以下异步函数添加到invoices.py文件中。

下面的代码使用AsyncIO来从一个主脚本管理多个脚本，如下一节所述。当您的项目扩展到包括多个API端点时，有一个良好的实践是让您的API消耗脚本拥有它们自己的文件。

# 使用异步函数包装代码，使其异步执行。
async def get_invoices():
	print("Going through the invoices")
	def main_request(baseurl,endpoint,x,headers):
	# 使用请求的get方法使用定义的参数拉取数据。
	# 使用f字符串在调用函数时迭代页面。
	r = requests.get(baseurl + endpoint + f'?page={x}',headers=headers)
	return r.json()

7. 最后，将以下代码添加到invoices.py文件中，其中get_pages函数处理API的分页。

此函数返回API中的总页面数，并帮助range函数迭代所有页面。

联系您的API提供商开发人员了解您的API提供商使用的分页方法。

  def get_pages(response):
			return response['meta']['total_pages']

	def parse_json(response):
	    charlist =[]
	    for item in response['invoices']:
	        charlist.append(item)
	    return charlist
	# 调用带有预定义参数的主函数。
	data = main_request(baseurl=baseurl,endpoint=endpoint,x=1,headers=headers)
	
	main_invoices = []
	# 通过API数据进行迭代/分页。
  # 您的API提供商可能使用不同的
	# 分页方法。
	for x in range(1,get_pages(data)+1):
	
	    print(x)
	    main_invoices.extend(parse_json(main_request(baseurl,endpoint,x,headers)))
	df_invoices = pd.DataFrame(main_invoices)
	
	# 将新数据写入SQL数据库。
	# 注意：确保数据库表名与此处提供的表名匹配。
	df_invoices.to_sql('invoices',con=engine,if_exists='replace',index=False)
	
  # 此函数等待get_invoices函数，
	# 在调用时运行get_invoices()函数，
	# 将被外部文件调用以调用您的API调用。

async def call_invoices():
	await get_invoices()

如果您希望向您的数据添加更多API端点：

重复“使用DBeaver管理MySQL数据库”部分的第四到第六步。

重复“获取和使用API数据”部分的所有步骤。

将API端点更改为您希望使用的另一个端点。

同步API端点

现在您有了数据库和API连接，可以通过运行invoices.py文件来开始API消费。但这样做将限制您同时使用一个API端点。

如何超越极限？您将创建另一个Python文件作为中心文件，该文件调用来自各个Python文件的API函数，并使用AsyncIO异步运行这些函数。这样，您可以保持编程的清晰性，并允许将多个函数捆绑在一起。

1. 创建一个名为central.py的新Python文件，并添加下面的代码。

类似于invoices.py文件，您将在每个步骤中向central.py文件添加代码片段，但是您可以在ATA的GitHub上查看完整的代码。

下面的代码导入了必要的模块，并使用from <filename> import <function name>语法从其他文件导入了脚本。

from datetime import datetime
from numpy import product
# 从invoices.py导入发票获取API函数
from invoices import call_invoices
import asyncio
import time
import schedule
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from datetime import datetime
import os

2. 接下来，在central.py文件中添加以下代码来控制来自invoices.py的脚本。

您需要引用/调用来自invoices.py的call_invoices函数到central.py中的AsyncIO任务（invoice_task）中。

# 创建主函数
async def chain():
	def _handle_task_result(task: asyncio.Task) -> None:
		try:
				task.result()
		except asyncio.CancelledError:
				pass
		except Exception:
				logging.exception('Exception raised by task %r',task)
	start = time.perf_counter()
	# 为从invoices.py导入的函数创建任务
	invoice_task = asyncio.create_task(call_invoices())

3. 创建AsyncIO任务后，等待任务以执行调用invoice.py中的call_invoices函数，一旦链式函数（在第二步中）开始运行。

	# 等待创建的任务
	await invoice_task
	# 跟踪任务的次数
	end = time.perf_counter()-start
	l_time =  time.localtime()
	human_time = time.asctime(l_time)
	print (f"chained result took {end:0.2f} seconds")
	print(f"Current time is {human_time}")

4. 创建一个AsyncIOScheduler来为脚本安排一个作业。此代码中添加的作业以一秒的间隔运行链式函数。

此作业很重要，以确保您的程序一直运行脚本以保持数据更新。

if name == "main":
# 调度程序使代码能够在循环中无限循环，并在循环之间休眠。
    scheduler = AsyncIOScheduler()
scheduler.add_job(chain,'interval',seconds=1)
scheduler.start()
print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

try:
	asyncio.get_event_loop().run_forever()
except (KeyboardInterrupt,SystemExit):
    pass