在 Cloud Storage 中查询
本指南介绍了数据在 Cloud Storage 中的存储方式，包括原始格式和处理后的格式。
原始数据
制造数据引擎 (MDE) 以原始格式存储收到的所有消息，而不进行处理。这对于归档和在出现配置或系统错误时重新处理数据都非常有用。
原始数据以 AVRO 格式存储，并使用以下架构：
{
"type" : "record",
"name" : "AvroPubsubMessageRecord",
"namespace" : "com.google.cloud.industry.manufacturing.sfp.datalake.core",
"fields" : [ {
"name" : "attributes",
"type" : {
"type" : "map",
"values" : "string"
}
}, {
"name" : "message",
"type" : {
"type" : "bytes",
"java-class" : "[B"
}
}, {
"name" : "messageId",
"type" : "string"
}, {
"name" : "timestamp",
"type" : "long"
} ]
该架构由以下值组成：
- attributes：存储 Pub/Sub 消息特性映射。
- message：存储在 Pub/Sub 主题
input-messages中收到的原始消息。 - messageId：Pub/Sub 在收到消息时设置的 ID，该 ID 会写入所有接收器，以便了解数据来源。
- 时间戳：Pub/Sub 收到消息的时间。
默认情况下，数据存储在名为 <project-id>-raw 的 Cloud Storage 存储桶中。然后，在版本文件夹（本例中为 1.3）下。
然后，使用日期分区按天分隔文件，格式为 dt=YYYY-MM-DD。
消息按 10 分钟的窗口分组，每个窗口中写入 10 个文件，这些数据用于生成文件名。以下脚本展示了这方面的一个示例：
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-00-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-01-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-02-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-03-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-04-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-05-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-06-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-07-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-08-of-10avro
gs://<project-id>-raw/v1.3/dt=2023-08-08/gcsoutput2023-08-08T09:20:00.000Z-2023-08-08T09:30:00.000Z-pane-0-last-09-of-10avro
从 BigQuery 中读取数据
以下部分介绍了从 BigQuery 读取数据的过程。
已处理的数据
MDE 处理数据后，可以配置将数据存储在 Cloud Storage 中。
默认情况下，数据会保存在名为 <project-id>-gcs-ingestion 的存储桶中。然后，它会存储在版本对应的文件夹下（在本例中为 v1.3），接着每种类型会存储在各自的文件夹 (default-discrete-records,
default-numeric-records, etc) 中，最后使用日期分区按天分隔文件，格式为 dt=YYYY-MM-DD。
与原始数据一样，消息按 10 分钟的时间窗口分组，每个窗口中写入 10 个文件，这些数据用于生成文件名，即：
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00000-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00001-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00002-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00003-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00004-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00005-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00006-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00007-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00008-of-00010.avro
gs://<project-id>-gcs-ingestion/v1.3/default-discrete-records/dt=2023-08-08/gcsoutput2023-08-08T09:40:00.000Z-2023-08-08T09:50:00.000Z-00009-of-00010.avro
每种原型都有不同的架构，但略有不同，因此每种类型都使用相应的架构。请参见以下示例：
NumericDataSeries 原型架构：
{
"type": "record",
"namespace": "com.google.cloud.industry.manufacturing.sfp.datalake.storage.gcs",
"name": "NumericDataSeriesGCSObject",
"fields": [
{
"name": "id",
"type": "string",
"doc": "Unique record id"
},
{
"name": "tag_name",
"type": "string",
"doc": "Name of the tag"
},
{
"name": "type_name",
"type": "string"
},
{
"name": "type_version",
"type": "int"
},
{
"name": "embedded_metadata",
"type": ["null", "string"],
"default": null
},
{
"name": "materialized_cloud_metadata",
"type": ["null", "string"],
"default": null
},
{
"name": "cloud_metadata_ref",
"type": ["null", "string"],
"default": null
},
{
"name": "source_message_id",
"type": "string"
},
{
"name": "event_timestamp",
"type": {
"type": "long",
"logicalType": "timestamp-millis"
}
},
{
"name": "value",
"type": "double"
}
]
}
DiscreteDataSeries 原型架构：
{
"type": "record",
"name": "DiscreteDataSeriesGCSObject",
"namespace": "com.google.cloud.industry.manufacturing.sfp.datalake.storage.gcs",
"fields": [
{
"name": "id",
"type": {
"type": "string",
"avro.java.string": "String"
},
"doc": "Unique record id"
},
{
"name": "tag_name",
"type": {
"type": "string",
"avro.java.string": "String"
},
"doc": "Name of the tag"
},
{
"name": "type_name",
"type": {
"type": "string",
"avro.java.string": "String"
}
},
{
"name": "type_version",
"type": "int"
},
{
"name": "embedded_metadata",
"type": [
"null",
{
"type": "string",
"avro.java.string": "String"
}
],
"default": null
},
{
"name": "materialized_cloud_metadata",
"type": [
"null",
{
"type": "string",
"avro.java.string": "String"
}
],
"default": null
},
{
"name": "cloud_metadata_ref",
"type": [
"null",
{
"type": "string",
"avro.java.string": "String"
}
],
"default": null
},
{
"name": "source_message_id",
"type": {
"type": "string",
"avro.java.string": "String"
}
},
{
"name": "event_timestamp",
"type": {
"type": "long",
"logicalType": "timestamp-millis"
}
},
{
"name": "data",
"type": {
"type": "string",
"avro.java.string": "String"
}
}
]
}
ContinuousDataSeries 原型架构：
{
"type": "record",
"namespace": "com.google.cloud.industry.manufacturing.sfp.datalake.storage.gcs",
"name": "ContinuousDataSeriesGCSObject",
"fields": [
{
"name": "id",
"type": "string",
"doc": "Unique record id"
},
{
"name": "tag_name",
"type": "string",
"doc": "Name of the tag"
},
{
"name": "type_name",
"type": "string"
},
{
"name": "type_version",
"type": "int"
},
{
"name": "embedded_metadata",
"type": ["null", "string"],
"default": null
},
{
"name": "materialized_cloud_metadata",
"type": ["null", "string"],
"default": null
},
{
"name": "cloud_metadata_ref",
"type": ["null", "string"],
"default": null
},
{
"name": "source_message_id",
"type": "string"
},
{
"name": "event_timestamp_start",
"type": {
"type": "long",
"logicalType": "timestamp-millis"
}
},
{
"name": "event_timestamp_end",
"type": {
"type": "long",
"logicalType": "timestamp-millis"
}
},
{
"name": "data",
"type": "string"
},
{
"name": "duration",
"type": "long"
}
]
}