python - 如何将 parquet 文件上传到 Azure ADLS 2 Blob

Question

您好，我想将 Parquet 文件上传到 ADLS gen 2 blob。我使用下面的代码行来创建 blob 并在其中上传 parquet 文件。

blob = BlobClient.from_connection_string(conn_str="Connection String", container_name="parquet", blob_name=outdir)
df.to_parquet('logs.parquet',compression='GZIP') #df is dataframe
with open("./logs.parquet", "rb") as data:
blob.upload_blob(data)
os.remove("logs.parquet")

我没有遇到任何错误，并且文件也写入 blob 中。但是，我认为我做得不对，因为 ADX/kusto 查询无法理解该文件，并且那里没有可见的数据。

以下是我在 Azure 数据资源管理器中执行的步骤，用于从 ADLS gen 2 中上传的 parquet 文件中获取记录。

已创建外部表:

.create external table LogDataParquet(AppId_s:string,UserId_g:string,Email_s:string,RoleName_s:string,Operation_s:string,EntityId_s:string,EntityType_s:string,EntityName_s:string,TargetTitle_s:string,TimeGenerated:datetime) 
kind=blob
dataformat=parquet
( 
   h@'https://streamoutalds2.blob.core.windows.net/stream-api-raw-testing;secret'
)
with 
(
   folder = "ExternalTables"   
)

外部表列映射:

.create external table LogDataParquet parquet mapping "LogDataMapparquet" '[{ "column" : "AppId_s", "path" : "$.AppId_s"},{ "column" : "UserId_g", "path" : "$"},{ "column" : "Email_s", "path" : "$.Email_s"},{ "column" : "RoleName_s", "path" : "$.RoleName_s"},{ "column" : "Operation_s", "path" : "$.Operation_s"},{ "column" : "EntityId_s", "path" : "$.EntityId_s"}]'

外部表未提供任何记录

external_table('LogDataParquet')

没有记录

external_table('LogDataParquet') | count 

1 条记录 - 计数 0

我使用流分析使用了类似的场景，我接收传入流并将其以 parquet 格式保存到 ADLS。在这种情况下，ADX 中的外部表可以很好地获取记录。我觉得我在 blob 中编写 Parquet 文件的方式犯了错误 - (使用 open("./logs.parquet", "rb") 作为数据:)

Answer 1

根据日志，外部表定义如下:

.create external table LogDataParquet(AppId_s:string,UserId_g:string,Email_s:string,RoleName_s:string,Operation_s:string,EntityId_s:string,EntityType_s:string,EntityName_s:string,TargetTitle_s:string,TimeGenerated:datetime) 
kind=blob
partition by 
   AppId_s,
   bin(TimeGenerated,1d)
dataformat=parquet
( 
   '******'
)
with 
(
   folder = "ExternalTables"   
)

PARTITION BY 子句告诉 ADX 预期的文件夹布局是:

<AppId_s>/<TimeGenerated, formatted as 'yyyy/MM/dd'>

例如:

https://streamoutalds2.blob.core.windows.net/stream-api-raw-testing;secret/SuperApp/2020/01/31

您可以在本节中找到有关 ADX 如何在查询期间在外部存储上查找文件的更多信息:https://learn.microsoft.com/en-us/azure/data-explorer/kusto/management/external-tables-azurestorage-azuredatalake#artifact-filtering-logic

要根据文件夹布局修复外部表定义，请使用.alter命令:

.alter external table LogDataParquet(AppId_s:string,UserId_g:string,Email_s:string,RoleName_s:string,Operation_s:string,EntityId_s:string,EntityType_s:string,EntityName_s:string,TargetTitle_s:string,TimeGenerated:datetime) 
kind=blob
dataformat=parquet
( 
  h@'https://streamoutalds2.blob.core.windows.net/stream-api-raw-testing;secret'
)
with 
(
   folder = "ExternalTables"   
)

顺便说一句，如果映射很简单(例如映射的列名称与数据源列名称匹配)，则 Parquet 格式不需要它。