将时序见解 (TSI) 第 2 代迁移到 Azure 数据资源管理器

概述

概略性迁移建议。

功能	第 2 代状态	建议的迁移
通过平展和转义从中心引入 JSON	TSI 引入	ADX - OneClick Ingest/向导
打开冷存储	客户存储帐户	连续数据导出（导出到 ADLS 中的客户指定外部表）。
PBI 连接器	个人预览版	使用 ADX PBI 连接器。 手动将 TSQ 重写为 KQL。
Spark 连接器	个人预览版。 查询遥测数据。 查询模型数据。	将数据迁移到 ADX。 将 ADX Spark 连接器用于遥测数据，将模型导出到 JSON 并加载到 Spark 中。 重写 KQL 中的查询。
批量上传	个人预览版	使用 ADX OneClick Ingest 和 LightIngest。 （可选）在 ADX 内设置分区。
时序模型		可以导出为 JSON 文件。 可以导入到 ADX，以便在 KQL 中进行联接。
TSI 资源管理器	进行冷暖切换	ADX 仪表板
查询语言	时序查询 (TSQ)	重写 KQL 中的查询。 使用 Kusto SDK 而不是 TSI SDK。

迁移遥测

使用存储帐户中的 PT=Time 文件夹检索环境中所有遥测的副本。 有关详细信息，请参阅数据存储。

迁移步骤 1 - 获取有关遥测数据的统计信息

数据

1. 环境概述
   • 记录数据访问 FQDN 第一部分中的环境 ID（例如，.env.crystal-dev.windows-int.net 中的 d390b0b0-1445-4c0c-8365-68d6382c1c2a）
2. 环境概述- > 存储配置- > 存储帐户
3. 使用存储资源管理器获取文件夹统计信息
   • 记录 PT=Time 文件夹的大小和 Blob 数。 对于批量导入个人预览版中的客户，还需记录 PT=Import 大小和 Blob 数。

迁移步骤 2 - 将遥测迁移到 ADX

创建 ADX 群集

1. 使用 ADX 成本估算器定义基于数据大小的群集大小。

   1. 从事件中心（或 IoT 中心）指标检索每日引入数据率。 从连接到 TSI 环境的存储帐户检索 TSI 使用的 Blob 容器中的数据量。 此信息将用于计算适用于环境的 ADX 群集的理想大小。
   2. 打开 Azure 数据资源管理器成本估算器，在现有字段中填充找到的信息。 将“工作负荷类型”设置为“存储优化”，在“热数据”中填充主动查询的数据总量。
   3. 在你提供所有信息后，Azure 数据资源管理器成本估算器会为群集建议一个 VM 大小和实例数。 分析主动查询数据的大小是否适合热缓存。 将建议的实例数乘以缓存大小或 VM 大小，例如：
      • 成本估算器建议：9x DS14 + 4 TB（缓存）
      • 建议的热缓存总计：36 TB = [9x（实例）x 4 TB（每个节点的热缓存）]
   4. 更多需考虑的因素：
      • 环境增长：规划 ADX 群集大小时，请考虑数据随时间的增长情况。
      • 混合和分区：在 ADX 群集中定义实例的数目时，请考虑使用额外的节点（原来的 2 到 3 倍），以加快混合和分区速度。
      • 有关计算选择的详细信息，请参阅为 Azure 数据资源管理器群集选择正确的计算 SKU。

2. 为了对群集和数据引入进行最佳监视，应启用诊断设置并将数据发送到 Log Analytics 工作区。

   1. 在 Azure 数据资源管理器边栏选项卡中转到“监视 | 诊断设置”，然后单击“添加诊断设置”

      

   2. 填写以下信息

      1. 诊断设置名称：此配置的“显示名称”
      2. 日志：至少选择 SucceededIngestion、FailedIngestion、IngestionBatching
      3. 选择要将数据发送到其中的 Log Analytics 工作区（如果没有该工作区，则需要在此步骤之前预配它）

      

3. 数据分区。

   1. 对于大多数数据集，默认的 ADX 分区就已足够。
   2. 数据分区有利于一组非常特定的场景，不得在其他场景中应用：
      1. 改进大数据集中的查询延迟，其中大多数查询会筛选高基数字符串列，例如时序 ID。
      2. 引入无序数据时，例如，过去事件可能在源中生成后的几天或几周内引入。
   3. 有关详细信息，请查看 ADX 数据分区策略。

准备进行数据引入

1. 转到 https://dataexplorer.azure.cn 。

   

2. 转到“数据”选项卡，然后选择“从 blob 容器引入”

   

3. 选择“群集”、“数据库”，然后使用为 TSI 数据选择的名称创建一个新表

   

4. 选择“下一步: 源”

5. 在“源”选项卡中，选择：

   1. 历史数据
   2. “选择容器”
   3. 为 TSI 数据选择订阅和存储帐户
   4. 选择与你的 TSI 环境关联的容器

   

6. 选择“高级设置”

   1. 创建时间模式：'/'yyyyMMddHHmmssfff'_'
   2. Blob 名称模式：*.parquet
   3. 选择“不等待引入完成”

   

7. 在“文件筛选器”下，添加 V=1/PT=Time 作为“文件夹路径”

   

8. 选择“下一步: 架构”

   注意

   在保存 Parquet 文件中的列时，TSI 会应用某些平展和转义。 有关更多详细信息，请参阅以下链接：平展和转义规则、引入规则更新。

• 如果架构未知或可变

  1. 删除不经常查询的所有列，至少保留时间戳和 TSID 列。

     

  2. 添加动态类型的新列，并使用 $ 路径将其映射到整个记录。

     

     示例：

     

• 如果架构已知或固定

  1. 确认数据看起来是正确的。 根据需要更正任何类型。
  2. 选择“下一步: 摘要”

复制 LightIngest 命令并将其存储在某个位置，以便在下一步使用它。

数据提取

在引入数据之前，需要安装 LightIngest 工具。 从 One-Click 工具生成的命令包含 SAS 令牌。 最好生成一个新的，以便控制到期时间。 在门户中导航到 TSI 环境的 Blob 容器，并选择“共享访问令牌”

1. 设置以下项

   1. 权限：“读取”和“列出”
   2. 到期时间：设置为足够数据完成迁移的一段时间

   

2. 选择“生成 SAS 令牌和 URL”并复制“Blob SAS URL”

   

3. 转到前面复制的 LightIngest 命令。 将命令中的 -source 参数替换为此“SAS Blob URL”

4. 选项 1：引入所有数据。 对于较小的环境，可以使用单个命令引入所有数据。

   1. 打开命令提示符，切换到已将 LightIngest 工具提取到其中的目录。 然后，粘贴并执行 LightIngest 命令。

   

5. 选项 2：按年份或月份引入数据。 如果环境较大，或者要对较小的数据集进行测试，可以进一步筛选 Lightingest 命令。

   1. 按年份：更改 -prefix 参数

      • 之前： -prefix:"V=1/PT=Time"
      • 之后： -prefix:"V=1/PT=Time/Y=<Year>"
      • 示例： -prefix:"V=1/PT=Time/Y=2021"

   2. 按月份：更改 -prefix 参数

      • 之前： -prefix:"V=1/PT=Time"
      • 之后： -prefix:"V=1/PT=Time/Y=<Year>/M=<month #>"
      • 示例： -prefix:"V=1/PT=Time/Y=2021/M=03"

修改该命令后，请执行它，如上所示。 （使用下面的监视选项）完成引入后，请针对要引入的下一个年份和月份修改命令。

监视引入

LightIngest 命令包括 -dontWait 标志，因此命令本身不会等待引入完成。 当引入正在进行时，监视进度的最佳方式是使用门户中的“见解”选项卡。 在门户中打开“Azure 数据资源管理器群集”部分，然后转到“监视 | 见解”

可以使用“引入(预览)”部分以及以下设置来监视正在进行的引入

• 时间范围：过去 30 分钟
• 查看“成功的引入”和“按表”查看
• 如果遇到任何故障，请查看“失败的引入”和“按表”查看

一旦看到表的指标为 0，就可以判断引入已完成。 若要查看更多详细信息，可以使用 Log Analytics。 在“Azure 数据资源管理器群集”部分，选择“日志”选项卡：

有用的查询

如果使用“动态架构”，则了解架构

| project p=treepath(fullrecord)
| mv-expand p 
| summarize by tostring(p)

访问数组中的值

| where id_string == "a"
| summarize avg(todouble(fullrecord.['nestedArray_v_double'])) by bin(timestamp, 1s)  
| render timechart 

将时序模型 (TSM) 迁移到 Azure 数据资源管理器

使用 TSI 资源管理器 UX 或 TSM 批处理 API，可以从 TSI 环境下载 JSON 格式的模型。 然后，可以将该模型导入到另一个系统，例如 Azure 数据资源管理器。

1. 从 TSI UX 下载 TSM。

2. 使用 VSCode 或另一个编辑器删除前三行。

   

3. 使用 VSCode 或另一个编辑器，搜索正则表达式 \},\n \{ 并将其替换为 }{

   

4. 使用“从文件上传”功能，将它作为 JSON 引入以独立表形式存在的 ADX。

   

将时序查询 (TSQ) 转换为 KQL

GetEvents

{
  "getEvents": {
    "timeSeriesId": [
      "assest1",
      "siteId1",
      "dataId1"
    ],
    "searchSpan": {
      "from": "2021-11-01T00:00:0.0000000Z",
      "to": "2021-11-05T00:00:00.000000Z"
    },
    "inlineVariables": {},
  }
}

events
| where timestamp >= datetime(2021-11-01T00:00:0.0000000Z) and timestamp < datetime(2021-11-05T00:00:00.000000Z)
| where assetId_string == "assest1" and siteId_string == "siteId1" and dataid_string == "dataId1"
| take 10000

带筛选器的 GetEvents

{
  "getEvents": {
    "timeSeriesId": [
      "deviceId1",
      "siteId1",
      "dataId1"
    ],
    "searchSpan": {
      "from": "2021-11-01T00:00:0.0000000Z",
      "to": "2021-11-05T00:00:00.000000Z"
    },
    "filter": {
      "tsx": "$event.sensors.sensor.String = 'status' AND $event.sensors.unit.String = 'ONLINE"
    }
  }
} 

events
| where timestamp >= datetime(2021-11-01T00:00:0.0000000Z) and timestamp < datetime(2021-11-05T00:00:00.000000Z)
| where deviceId_string== "deviceId1" and siteId_string == "siteId1" and dataId_string == "dataId1"
| where ['sensors.sensor_string'] == "status" and ['sensors.unit_string'] == "ONLINE"
| take 10000

带投影变量的 GetEvents

{
  "getEvents": {
    "timeSeriesId": [
      "deviceId1",
      "siteId1",
      "dataId1"
    ],
    "searchSpan": {
      "from": "2021-11-01T00:00:0.0000000Z",
      "to": "2021-11-05T00:00:00.000000Z"
    },
    "inlineVariables": {},
    "projectedVariables": [],
    "projectedProperties": [
      {
        "name": "sensors.value",
        "type": "String"
      },
      {
        "name": "sensors.value",
        "type": "bool"
      },
      {
        "name": "sensors.value",
        "type": "Double"
      }
    ]
  }
}	 

events
| where timestamp >= datetime(2021-11-01T00:00:0.0000000Z) and timestamp < datetime(2021-11-05T00:00:00.000000Z)
| where deviceId_string== "deviceId1" and siteId_string == "siteId1" and dataId_string == "dataId1"
| take 10000
| project timestamp, sensorStringValue= ['sensors.value_string'], sensorBoolValue= ['sensors.value_bool'], sensorDoublelValue= ['sensors.value_double']

AggregateSeries

{
  "aggregateSeries": {
    "timeSeriesId": [
      "deviceId1"
    ],
    "searchSpan": {
      "from": "2021-11-01T00:00:00.0000000Z",
      "to": "2021-11-05T00:00:00.0000000Z"
    },
    "interval": "PT1M",
    "inlineVariables": {
      "sensor": {
        "kind": "numeric",
        "value": {
          "tsx": "coalesce($event.sensors.value.Double, todouble($event.sensors.value.Long))"
        },
        "aggregation": {
          "tsx": "avg($value)"
        }
      }
    },
    "projectedVariables": [
      "sensor"
    ]
  }	

events
| where timestamp >= datetime(2021-11-01T00:00:00.0000000Z) and timestamp < datetime(2021-11-05T00:00:00.0000000Z)
| where  deviceId_string == "deviceId1"
| summarize avgSensorValue= avg(coalesce(['sensors.value_double'], todouble(['sensors.value_long']))) by bin(IntervalTs = timestamp, 1m)
| project IntervalTs, avgSensorValue

带筛选器的 AggregateSeries

{
  "aggregateSeries": {
    "timeSeriesId": [
      "deviceId1"
    ],
    "searchSpan": {
      "from": "2021-11-01T00:00:00.0000000Z",
      "to": "2021-11-05T00:00:00.0000000Z"
    },
    "filter": {
      "tsx": "$event.sensors.sensor.String = 'heater' AND $event.sensors.location.String = 'floor1room12'"
    },
    "interval": "PT1M",
    "inlineVariables": {
      "sensor": {
        "kind": "numeric",
        "value": {
          "tsx": "coalesce($event.sensors.value.Double, todouble($event.sensors.value.Long))"
        },
        "aggregation": {
          "tsx": "avg($value)"
        }
      }
    },
    "projectedVariables": [
      "sensor"
    ]
  }
}	

events
| where timestamp >= datetime(2021-11-01T00:00:00.0000000Z) and timestamp < datetime(2021-11-05T00:00:00.0000000Z)
| where  deviceId_string == "deviceId1"
| where ['sensors.sensor_string'] == "heater" and ['sensors.location_string'] == "floor1room12"
| summarize avgSensorValue= avg(coalesce(['sensors.value_double'], todouble(['sensors.value_long']))) by bin(IntervalTs = timestamp, 1m)
| project IntervalTs, avgSensorValue

从 TSI Power BI 连接器迁移到 ADX Power BI 连接器

此迁移涉及的手动步骤如下

1. 将 Power BI 查询转换为 TSQ
2. 将 TSQ 转换为 KQL Power BI 查询：从 TSI UX 资源管理器复制的 Power BI 查询如下所示

对于原始数据 (GetEvents API)

{"storeType":"ColdStore","isSearchSpanRelative":false,"clientDataType":"RDX_20200713_Q","environmentFqdn":"6988946f-2b5c-4f84-9921-530501fbab45.env.timeseries.azure.com", "queries":[{"getEvents":{"searchSpan":{"from":"2019-10-31T23:59:39.590Z","to":"2019-11-01T05:22:18.926Z"},"timeSeriesId":["Arctic Ocean",null],"take":250000}}]}

• 若要将其转换为 TSQ，请从上述有效负载生成 JSON。 GetEvents API 文档还提供可以更好地了解它的示例。 查询 - 执行 - REST API（Azure 时序见解）| Microsoft Docs
• 转换后的 TSQ 如下所示。 它是“查询”中的 JSON 有效负载

{
  "getEvents": {
    "timeSeriesId": [
      "Arctic Ocean",
      "null"
    ],
    "searchSpan": {
      "from": "2019-10-31T23:59:39.590Z",
      "to": "2019-11-01T05:22:18.926Z"
    },
    "take": 250000
  }
}

对于聚合数据（聚合系列 API）

• 对于单一内联变量，来自 TSI UX 资源管理器的 PowerBI 查询如下所示：

{"storeType":"ColdStore","isSearchSpanRelative":false,"clientDataType":"RDX_20200713_Q","environmentFqdn":"6988946f-2b5c-4f84-9921-530501fbab45.env.timeseries.azure.com", "queries":[{"aggregateSeries":{"searchSpan":{"from":"2019-10-31T23:59:39.590Z","to":"2019-11-01T05:22:18.926Z"},"timeSeriesId":["Arctic Ocean",null],"interval":"PT1M", 		"inlineVariables":{"EventCount":{"kind":"aggregate","aggregation":{"tsx":"count()"}}},"projectedVariables":["EventCount"]}}]}

• 若要将其转换为 TSQ，请从上述有效负载生成 JSON。 AggregateSeries API 文档还提供可以更好地了解它的示例。 查询 - 执行 - REST API（Azure 时序见解）| Microsoft Docs
• 转换后的 TSQ 如下所示。 它是“查询”中的 JSON 有效负载

{
  "aggregateSeries": {
    "timeSeriesId": [
      "Arctic Ocean",
      "null"
    ],
    "searchSpan": {
      "from": "2019-10-31T23:59:39.590Z",
      "to": "2019-11-01T05:22:18.926Z"
    },
    "interval": "PT1M",
    "inlineVariables": {
      "EventCount": {
        "kind": "aggregate",
        "aggregation": {
          "tsx": "count()"
        }
      }
    },
    "projectedVariables": [
      "EventCount",
    ]
  }
}

• 对于多个内联变量，请将 json 追加到“inlineVariables”中，如下面的示例所示。 多个内联变量的 Power BI 查询如下所示：

{"storeType":"ColdStore","isSearchSpanRelative":false,"clientDataType":"RDX_20200713_Q","environmentFqdn":"6988946f-2b5c-4f84-9921-530501fbab45.env.timeseries.azure.com","queries":[{"aggregateSeries":{"searchSpan":{"from":"2019-10-31T23:59:39.590Z","to":"2019-11-01T05:22:18.926Z"},"timeSeriesId":["Arctic Ocean",null],"interval":"PT1M", "inlineVariables":{"EventCount":{"kind":"aggregate","aggregation":{"tsx":"count()"}}},"projectedVariables":["EventCount"]}}, {"aggregateSeries":{"searchSpan":{"from":"2019-10-31T23:59:39.590Z","to":"2019-11-01T05:22:18.926Z"},"timeSeriesId":["Arctic Ocean",null],"interval":"PT1M", "inlineVariables":{"Magnitude":{"kind":"numeric","value":{"tsx":"$event['mag'].Double"},"aggregation":{"tsx":"max($value)"}}},"projectedVariables":["Magnitude"]}}]}

{
  "aggregateSeries": {
    "timeSeriesId": [
      "Arctic Ocean",
      "null"
    ],
    "searchSpan": {
      "from": "2019-10-31T23:59:39.590Z",
      "to": "2019-11-01T05:22:18.926Z"
    },
    "interval": "PT1M",
    "inlineVariables": {
      "EventCount": {
        "kind": "aggregate",
        "aggregation": {
          "tsx": "count()"
        }
      },
      "Magnitude": {
        "kind": "numeric",
        "value": {
          "tsx": "$event['mag'].Double"
        },
        "aggregation": {
          "tsx": "max($value)"
        }
      }
    },
    "projectedVariables": [
      "EventCount",
      "Magnitude",
    ]
  }
}

• 如果要查询最新的数据 ("isSearchSpanRelative": true)，请按如下所述手动计算 searchSpan
  • 从 Power BI 有效负载中找出 "from" 和 "to" 之间的差异。 让我们将该差异称之为 "D"，其中 "D" = "from" - "to"
  • 采用当前时间戳 ("T") 并减去在第一步获得的差异。 它将是 searchSpan 的新 "from" (F)，其中 "F" = "T" - "D"
  • 现在，在步骤 2 中获取的新 "from" 为 "F"，新 "to" 为 "T"（当前时间戳）