从 SELECT 语句定义 EXTRACT 范围

Question

我打算批量处理存储在ADLA中的EventHub中的数据集。在我看来，处理间隔似乎是合乎逻辑的，其中我的日期介于上次执行日期时间和当前执行日期时间之间。

我想过将执行时间戳保存在表中，以便可以跟踪它，并执行以下操作：

DECLARE @my_file string = @"/data/raw/my-ns/my-eh/{date:yyyy}/{date:MM}/{date:dd}/{date:HH}/{date:mm}/{date:ss}/{*}.avro";
DECLARE @max_datetime DateTime =  DateTime.Now;
@min_datetime =
    SELECT (DateTime) MAX(execution_datetime) AS min_datetime
    FROM my_adldb.dbo.watermark;
@my_json_bytes =
    EXTRACT Body byte[],
            date DateTime
    FROM @my_file
    USING new Microsoft.Analytics.Samples.Formats.ApacheAvro.AvroExtractor(@"{""type"":""record"",""name"":""EventData"",""namespace"":""Microsoft.ServiceBus.Messaging"",""fields"":[{""name"":""SequenceNumber"",""type"":""long""},{""name"":""Offset"",""type"":""string""},{""name"":""EnqueuedTimeUtc"",""type"":""string""},{""name"":""SystemProperties"",""type"":{""type"":""map"",""values"":[""long"",""double"",""string"",""bytes""]}},{""name"":""Properties"",""type"":{""type"":""map"",""values"":[""long"",""double"",""string"",""bytes"",""null""]}},{""name"":""Body"",""type"":[""null"",""bytes""]}]}");

如何将此间隔正确添加到我的EXTRACT查询中？我使用手动定义的间隔的通用 WHERE 子句对其进行了测试，它有效，但是当我尝试使用 @min_datetime 时它不起作用，因为它的结果是一个行集。

我考虑过在后续查询中应用一些过滤，但恐怕这意味着@my_json_bytes将提取我的整个数据集并对其进行过滤，从而导致查询欠优化。

提前谢谢。

Answer 1

您应该

能够应用过滤器作为以后SELECT的一部分。 U-SQL 可以在某些情况下向上推送谓词，但我还没有能够测试这一点。 尝试这样的事情：

@min_datetime =
    SELECT (DateTime) MAX(execution_datetime) AS min_datetime
    FROM my_adldb.dbo.watermark;
@my_json_bytes =
    EXTRACT Body byte[],
            date DateTime
    FROM @my_file
    USING new Microsoft.Analytics.Samples.Formats.ApacheAvro.AvroExtractor(@"{""type"":""record"",""name"":""EventData"",""namespace"":""Microsoft.ServiceBus.Messaging"",""fields"":[{""name"":""SequenceNumber"",""type"":""long""},{""name"":""Offset"",""type"":""string""},{""name"":""EnqueuedTimeUtc"",""type"":""string""},{""name"":""SystemProperties"",""type"":{""type"":""map"",""values"":[""long"",""double"",""string"",""bytes""]}},{""name"":""Properties"",""type"":{""type"":""map"",""values"":[""long"",""double"",""string"",""bytes"",""null""]}},{""name"":""Body"",""type"":[""null"",""bytes""]}]}");
@working =
    SELECT *
    FROM @my_json_bytes AS j
         CROSS JOIN
             @min_datetime AS t
    WHERE j.date > t.min_datetime;