最优ETL流程和平台

Question

我面临以下问题，我是云计算和数据库的新手。我想为应用程序设置一个简单的仪表板。基本上我想复制这个显示空气污染数据的网站。 https://airtube.info/

在我看来，我需要做的事情如下：

1. 从 API 下载数据：https://github.com/opendata-stuttgart/meta/wiki/EN-APIs，我特别想到了这个链接“https://data.sensor.community/static/v2/data.1h.json - 过去一小时每个传感器的所有测量值的平均值。” （技术：Python 机器人）
2. 设置一个机器人来稍微转换数据以根据我们的需要定制它们。 （技术：Python）
3. 将数据上传到数据库。 （技术：Google Big-Query 或 AWS）
4. 将数据库连接到可视化工具，这样每个人都可以在我们的网页上看到它。 （技术：可能是 Python 中的 Dash）

我的问题如下。 1. 你是否同意我的想法，或者你会改变一些元素以使其更有效率？ 2. 你如何看待运行 python 脚本来转换数据？有没有更简单的想法？ 3. 您建议使用哪种技术来建立数据库？

感谢cmets！ 此致， 巴特克

Answer 1

如果您想对您的数据进行一些分析，我建议您将数据上传到 BigQuery，完成后，您可以在此处创建新查询并获得您想要分析的结果。我正在检查数据集“data.1h.json”，我会使用这样的架构在 BigQuery 中创建一个表：

CREATE TABLE dataset.pollution
(
  id NUMERIC,
  sampling_rate STRING,
  timestamp TIMESTAMP,
  location STRUCT<
  id NUMERIC,
  latitude FLOAT64,
  longitude FLOAT64,
  altitude FLOAT64,
  country STRING,
  exact_location INT64,
  indoor INT64
  >,
  sensor STRUCT<
    id NUMERIC,
    pin STRING,
    sensor_type STRUCT<
      id INT64,
      name STRING,
      manufacturer STRING
    >
  >,
  sensordatavalues ARRAY<STRUCT<
    id NUMERIC,
    value FLOAT64,
    value_type STRING
  >>
)

好的，我们已经创建了表，所以现在我们需要将 JSON 文件中的所有数据插入到该表中，并且由于您想使用 Python，我将使用 BigQuery Python 客户端库 [1 ] 从必须存储文件的 Google Cloud Storage [2] 中的存储桶中读取数据，并转换数据以将其上传到 BigQuery 表。

代码是这样的：

from google.cloud import storage
import json
from google.cloud import bigquery

client = bigquery.Client()

table_id = "project.dataset.pollution"
# Instantiate a Google Cloud Storage client and specify required bucket and 
file
storage_client = storage.Client()
bucket = storage_client.get_bucket('bucket')
blob = bucket.blob('folder/data.1h.json')
table = client.get_table(table_id)
# Download the contents of the blob as a string and then parse it using 
json.loads() method
data = json.loads(blob.download_as_string(client=None))

# Partition the request in order to avoid reach quotas
partition = len(data)/4

cont = 0
data_aux = []
for part in data:
    if cont >= partition:
        errors = client.insert_rows(table, data_aux)  # Make an API request.
        if errors == []:
            print("New rows have been added.")
        else:
            print(errors)
        cont = 0
        data_aux = []
    # Avoid empty values (clean data)
    if part['location']['altitude'] is "":
        part['location']['altitude'] = 0
    if part['location']['latitude'] is "":
        part['location']['latitude'] = 0
    if part['location']['longitude'] is "":
        part['location']['longitude'] = 0
    data_aux.append(part)
    cont += 1

正如您在上面看到的，我必须创建一个分区以避免达到请求大小的配额。在这里您可以看到要避免的配额数量 [3]。

另外，位置字段中的一些Data似乎有空值，所以需要对其进行控制以避免错误。

由于您已经将数据存储在 BigQuery 中，为了创建新的仪表板，我将使用 Data Studio 工具 [4] 来可视化您的 BigQuery 数据并针对您要显示的列创建查询。

[1]https://cloud.google.com/bigquery/docs/reference/libraries#using_the_client_library

[2]https://cloud.google.com/storage

[3]https://cloud.google.com/bigquery/quotas

[4]https://cloud.google.com/bigquery/docs/visualize-data-studio