Azure Blob 文件索引检索查询 -- Azure Blob Indexing 篇

前面我们为大家介绍了 Azure Blob 文件索引检索查询的几个典型场景，本文为大家来介绍一下 Azure Blob Indexing 的方案实现。我们来回顾一下架构图，在 Azure Blob 中原生提供了 Blob Indexing 功能，支持典型的 Key Value 检索查询，如果用户对于元数据的检索查询需求是完全匹配，Blob Indexing 是一个非常好的方式，客户不需要额外借助其他外部系统来构建文件查询索引。虽然 Blob Indexing 的方式虽然简单易用，但也有场景的限制，比如上面所述的模糊查询是无法通过 Blob Indexing 来做到的，目前 Blob Indexing 对于构建的 Key Value 都是按照 String 字符组数据类型来处理的，在查询检索时除支持等于匹配外也支持大于小于等匹配符，但由于 Value 按照字符串存储，所以对于大小匹配符的支持按照字母排序来进行比较，对于数字，日期等类型数据也是按照字符串来处理，所以对于大小匹配符的支持就会受限，同时对于数字类型数据的比较只支持第一个数字的比较（比如 11 和 12 比较，> 11 结果中不会有 12），如果用户有数字比较的场景，建议客户可以把数字拆分成多个0-9 位，创建多个 Key Value 索引来实现（需要注意目前 Azure Blob 单个文件上最多支持 10 个 Key Value）。所以如果用户希望对元数据做模糊查询，可以考虑通过 RedisSearch/Elastic/Azure Search 来构建外部查询索引，借助外部索引强大的搜索能力来进行支持，其中 RedisSearch 和 Azure Research 微软提供第一方的 PaaS 服务，可以方便用户快速构建查询索引平台。Azure Search 作为微软自研的搜索产品，其对于微软第一方的服务原生集成是最大的优势，客户可以通过简单的点击即可将 Blob 的数据信息导入到 Azure Search 中，另外 Azure Search 内的 AI 赋能可以增加对文件中富媒体内容的支持，这部分我们放在后面文件内容检索查询部分给大家单独介绍。RedisSearch/ElasticSearch 都是开源的索引查询产品，其中 RedisSearch 在微软云上的 Redis Enterprise Tier 已经原生支持，客户可以通过托管 PaaS 的方式快速创建使用，ElasticSearch 客户可以通过 Marketplace 购买使用。无论时 Azure Blob Indexing 还是外部查询索引系统（RedisSearch/ElasticSearch），用户都需要来实时的维护索引信息，当 Blob 文件进行增删改动作时，相应的需要对索引数据进行变更。整个系统架构的实现可以通过 EventGrid 来监听 Blob 的增删改动作，然后触发代码来进行索引的更新。对于索引变更的管理逻辑代码可以通过托管在 Azure Function 服务中，快速的实现事件触发的逻辑。用户侧的查询可以通过 Webapp 构建前端 GUI 层为用户提供访问。

目前 Azure Blob Indexing 功能处于 Preview 预览阶段，感兴趣的用户可以在全球版 Azure 的 Canada Central，Canada East, France Central 和 France South 创建。创建时用户需要选择 General Purpose V2 Storage Account（且不开启 hierachical namespace），具体创建过程大家可以访问链接，这里我们就不做赘述。在整个方案中，我们需要通过代码来维护 Blob Indexing 中的索引数据，下面我们来看一下代码的实现。

数据准备：示例中通过本地的书籍信息样本数据为大家进行介绍，原数据是一个 CSV 文件，文件中包含如下书籍信息

运行环境准备：示例中采用 python 作为示例代码，客户需要环境中安装 azure-storage-blob 依赖包，并从 Azure Portal 中读取 Blob 服务的链接字符串

from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
import csv

AZURE_STORAGE_CONNECTION_STRING="Input_your_blob_connectionstring_here"

# Create Blob Client
blob_service_client = BlobServiceClient.from_connection_string(AZURE_STORAGE_CONNECTION_STRING)

# Create a unique name for the container
container_name = "indexdemo"

# Create the container
container_client = blob_service_client.create_container(container_name)

数据索引：通过读取 CSV file 中的数据信息，仿真创建 Blob 并将作者，书名，发行商和发行时间作为索引数据。

data = "demo"

with open('demobookdataset.csv', mode='r') as csv_file:
    csv_reader = csv.DictReader(csv_file)
    line_count = 0
    for row in csv_reader:
        if line_count == 0:
            print(f'Column names are {", ".join(row)}')
            line_count += 1
            continue

        # Create a blob client using the local file name as the name for the blob
        blob_client = blob_service_client.get_blob_client(container=container_name, blob=str(line_count))

        # Upload blob Data
        blob_client.upload_blob(data)

        # Preparing Index Key Value Data
        tagmark = {}
        tagmark["author"] = row["authors"]
        tagmark["title"] = row["title"]
        tagmark["publisher"] = row["publisher"]
        tagmark["publicationdate"] = row["publication_date"]

        # Set Blob Indexing
        blob_client.set_blob_tags(tagmark)
        line_count += 1

数据检索查询：示例中通过以索引中 author 作为查询值进行检索查询验证，需要注意在 Filter 的查询语句编写语法上需要按照规范进行编写，可以参考查询语法进行定义。

# Full loop all the documents index
with open('demobookdataset.csv', mode='r') as csv_file:
    csv_reader = csv.DictReader(csv_file)
    line_count = 0
    for row in csv_reader:
        if line_count == 0:
            #print(f'Column names are {", ".join(row)}')
            line_count += 1
            continue
        filter = ""author" = '" + row["authors"] + "'"
        result = blob_service_client.find_blobs_by_tags(filter)
        for item in result:
            print(item)

通过上述示例我们了解到 Blob Indexing 的代码实现，大家可以按照自己喜好的开发语言进行开发，整个方案关于 EventGrid， Function 的集成这里不再赘述，感兴趣的小伙伴，可以参考我之前的博客Azure NSG Flow Log 引发的自嗨。