Azure Blob存储/ NodeJS -从缓冲区读取Avro文件
Azure Blob存储/ NodeJS -从缓冲区读取Avro文件
提问于 2021-01-01 11:22:30
我已经让事件集线器使用捕获功能输出到Blob Storage -这会将我们抛入队列的所有内容输出为.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);输出到控制台时的缓冲区代码段：
<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 (我不想在内容上做一些假设，试图把消息正文拉出来)。
有没有人成功地从基于缓冲区的Azure中提升了.avro内容？我在网上看到了很多关于将这些文件加载到Spark或Kafka中的指导，但不仅仅是读取流中的文件。
谢谢!
回答 1
Stack Overflow用户
回答已采纳
发布于 2021-01-04 11:14:58
对于这个问题，我们可以使用avsc包来解析带有buffer的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();
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/65527093
复制相关文章
相似问题
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号