从 U-SQL 上的特定文件读取

Question

我是U-SQL的新手，已经开始查询文件。根据我在文档和堆栈溢出中看到的说明，我编写了一个查询，用于从一组 json 文件中提取元数据，如下所示。

REFERENCE ASSEMBLY [Newtonsoft.Json];
REFERENCE ASSEMBLY [Microsoft.Analytics.Samples.Formats];
USING Microsoft.Analytics.Samples.Formats.Json;
DECLARE @InputFile string = "/test/{*}.json";
DECLARE @OutputFile string = "/metadata.csv";
@json =
EXTRACT
id string,
date DateTime,
type string,
uri = FILE.URI()
, modified_date = FILE.MODIFIED()
, created_date = FILE.CREATED()
, file_sz = FILE.LENGTH()
FROM
@InputFile
USING new JsonExtractor();
@json2 =
SELECT 
uri
modified_date,
created_date,
file_sz
FROM @json;
@json3 =
SELECT DISTINCT uri,
modified_date,
created_date,
file_sz
FROM @json2;
OUTPUT @json3
TO @OutputFile
USING Outputters.Csv(outputHeader:true,quoting:true);
DROP ASSEMBLY [Newtonsoft.Json];
DROP ASSEMBLY [Microsoft.Analytics.Samples.Formats];

这将生成所需的元数据(我在 Azure 门户上运行它，即使这里的位置是相对的(

我的问题是：

1( 如何使用内部表/文件中的列(文件名列(中的值作为输入的文件路径列表？

2(我们如何将新数据附加到现有文件并使用新文件列表更新元数据文件。

我的元数据如下所示：

uri           created_date       modified_date   file_sz
/…/abc.json      09-22-2018        09-23-2018       250
/…/del.json      09-24-2018        09-24-2018       126

预期输出(如果可能(

@filespresent =
SELECT uri
FROM @metadata
DECLARE @Inputfile string = @filespresent

这样做的主要目的是我每天都会收到一批新文件，我只想读取今天上传的文件。

文件名不包含日期，我可以提取日期信息的唯一方法是从文件内部。我在单独的查询中提取元数据，然后对从元数据文件中选择的文件运行主作业。

如果有其他解决方法，也非常欢迎。

任何帮助，不胜感激。

Answer 1

您想要的功能(例如，从文件中读取路径并附加到现有文件(不可用。

追加到文件时，您可以通过读取文件、合并新数据，然后将结果写入同一文件来完成。

但是，通过查看您的方案 **这样做的主要目的是我每天都会收到一批新文件，我只想读取今天上传的文件。

您可以按上述方式执行EXTRACT，然后在created_date或modified_date列上放置过滤器，以仅选择为特定日期创建或修改的文件。 例如(键入堆栈溢出(，

DECLARE EXTERNAL @last_processed_modified_date = DateTime.Now();
@json = EXTRACT
id string,
date DateTime,
type string,
uri = FILE.URI()
, modified_date = FILE.MODIFIED()
, created_date = FILE.CREATED()
, file_sz = FILE.LENGTH()
FROM @InputFile
USING new JsonExtractor();
@json = SELECT * FROM @json WHERE modified_date > @last_processed_modified_date;
…

Answer 2

这是目前的限制。您可以在此处为该功能投票。这里也说了同样的问题。

解决此问题的唯一方法是使用注入外部变量的powershell运行u-sql脚本，如下所示：

DECLARE EXTERNAL @InputFile string = "/Input/sample{n}.json";

如果您尝试运行它，则会收到"行集变量不是标量变量"的错误...

REFERENCE ASSEMBLY [Newtonsoft.Json];
REFERENCE ASSEMBLY [Microsoft.Analytics.Samples.Formats];
USING Microsoft.Analytics.Samples.Formats.Json;
DECLARE @InputFile string = "/Input/sample4.json";
DECLARE @OutputFile1 string = "/Output/metadata1.csv";
DECLARE @OutputFile2 string = "/Output/metadata2.csv";
DECLARE @OutputFile3 string = "/Output/metadata3.csv";
@json1 =
EXTRACT
// Json fields
id string,   
file string,
// Virtual columns
n string,
uri = FILE.URI()
, modified_date = FILE.MODIFIED()
, created_date = FILE.CREATED()
, file_sz = FILE.LENGTH()
FROM
@InputFile // This file has a file set within
USING new JsonExtractor();
OUTPUT @json1
TO @OutputFile1
USING Outputters.Csv(outputHeader:true,quoting:true);
@json2 =
SELECT 
file
FROM @json1;
OUTPUT @json2
TO @OutputFile2
USING Outputters.Csv(outputHeader:true,quoting:true);
@json3 =
EXTRACT
// Json fields
id string,   
file string,
// Virtual columns
n string,
uri = FILE.URI()
, modified_date = FILE.MODIFIED()
, created_date = FILE.CREATED()
, file_sz = FILE.LENGTH()
FROM
@json2 // This is a rowset variable (with our fileset)
USING new JsonExtractor();
OUTPUT @json3
TO @OutputFile3
USING Outputters.Csv(outputHeader:true,quoting:true);