Azure Blob 存储/NodeJS - 从缓冲区读取 Avro 文件

Question

我已经使用捕获功能将事件中心输出到 Blob 存储 - 这会将我们放入队列中的任何内容作为 .avro 文件输出。

如果我下载此文件并尝试使用 avro-js 之类的库对其进行解析，我没有问题 - 我可以很好地读取文件并按照我认为合适的方式处理内容。

但是 - 在使用 Node 处理 Azure Blob 存储时，我希望将文件处理为已下载的文件。读取文件时返回的格式是缓冲区，但我找不到使用库成功解析此文件的方法（找不到正确的方法，如果有的话）。

用于从 Azure 下载 blob 的代码，省略了一些位：

const { BlobServiceClient } = require('@azure/storage-blob');
const blobServiceClient = BlobServiceClient.fromConnectionString(AZURE_STORAGE_CONNECTION_STRING);
const containerClient = blobServiceClient.getContainerClient("data");

const blockBlobClient = containerClient.getBlockBlobClient(blob.name);
    
const downloadBlockBlobResponse = await blockBlobClient.download(0);

输出到控制台时缓冲sn-p：

<Buffer 4f 62 6a 01 04 14 61 76 72 6f 2e 63 6f 64 65 63 08 6e 75 6c 6c 16 61 76 72 6f 2e 73 63 68 65 6d 61 ec 06 7b 22 74 79 70 65 22 3a 22 72 65 63 6f 72 64 ... 589 more bytes>

转成字符串时的内容（粘贴图片时乱码输出不正确）：

已尝试将 .avro 文件作为纯文本读取，虽然它们大多没问题，但有一些字符是乱码，因此它不会读取为 JSON（我不想做出假设在内容上尝试将消息正文拉出）。

有没有人基于 Buffers 从 Azure 成功提取 .avro 内容？ 我在网上看到很多关于将这些加载到 Spark 或 Kafka 中的指导，但不仅仅是读取流中的文件。

谢谢！

Answer 1

关于这个问题，我们可以使用包avsc来解析带有缓冲区的avro文件。更多详情请参考here。

例如

const avro = require("avsc");
const {
  BlobServiceClient,
  StorageSharedKeyCredential,
} = require("@azure/storage-blob");

const accountName = "";
const accountKey =
  "";
async function main() {
  const creds = new StorageSharedKeyCredential(accountName, accountKey);
  const blobServiceClient = new BlobServiceClient(
    `https://${accountName}.blob.core.windows.net`,
    creds
  );
  const containerClient = blobServiceClient.getContainerClient("");
  const blockBlobClient = containerClient.getBlockBlobClient(
    ""
  );

  const downloadBlockBlobResponse = await blockBlobClient.download(0);
  const buf = await streamToBuffer(
    downloadBlockBlobResponse.readableStreamBody
  );

  const decoder = new avro.streams.BlockDecoder({
    parseHook: (schema) => {
      console.log("the avro file schema:");
      console.log(schema);
      return avro.Type.forSchema(schema, { wrapUnions: true });
    },
  });

  decoder.on("data", (data) => {
    console.log(data);
  });
  decoder.end(buf);
}
async function streamToBuffer(readableStream) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    readableStream.on("data", (data) => {
      chunks.push(data instanceof Buffer ? data : Buffer.from(data));
    });
    readableStream.on("end", () => {
      resolve(Buffer.concat(chunks));
    });
    readableStream.on("error", reject);
  });
}
main();