U-SQL根据文件中的Distinct值将CSV文件拆分为多个文件

Question

我在Azure Data Lake Store中有数据，我正在使用U-SQL处理Azure Data Analytic Job中存在的数据。我有包含空间数据的几个CSV文件，与此类似：

File_20170301.csv

longtitude| lattitude | date   | hour | value1 
    ----------+-----------+--------------+------+------- 
    45.121 | 21.123 | 2017-03-01 | 01 | 20 
    45.121 | 21.123 | 2017-03-01 | 02 | 10 
    45.121 | 21.123 | 2017-03-01 | 03 | 50 
    48.121 | 35.123 | 2017-03-01 | 01 | 60 
    48.121 | 35.123 | 2017-03-01 | 02 | 15 
    48.121 | 35.123 | 2017-03-01 | 03 | 80 

File_20170302.csv

longtitude| lattitude | date   | hour | value1 
    ----------+-----------+--------------+------+------- 
    45.121 | 21.123 | 2017-03-02 | 01 | 20 
    45.121 | 21.123 | 2017-03-02 | 02 | 10 
    45.121 | 21.123 | 2017-03-02 | 03 | 50 
    48.121 | 35.123 | 2017-03-02 | 01 | 60 
    48.121 | 35.123 | 2017-03-02 | 02 | 15 
    48.121 | 35.123 | 2017-03-02 | 03 | 80 

每个文件包含的数据不同的日期和所有经度 - 组合。

我想合并我所有的文件并拆分数据，这样我就可以为每个经度 - 折点组合结束一个文件。

所以，通过我的文件夹中的所有文件循环和附加的所有数据全部天后，我将结束与以下：

File_45_21.csv

longtitude| lattitude | date   | hour | value1 
    ----------+-----------+--------------+------+------- 
    45.121 | 21.123 | 2017-03-01 | 01 | 20 
    45.121 | 21.123 | 2017-03-01 | 02 | 10 
    45.121 | 21.123 | 2017-03-01 | 03 | 50 
    45.121 | 21.123 | 2017-03-02 | 01 | 20 
    45.121 | 21.123 | 2017-03-02 | 02 | 10 
    45.121 | 21.123 | 2017-03-02 | 03 | 50 

File_48_35.csv

longtitude| lattitude | date   | hour | value1 
    ----------+-----------+--------------+------+------- 
    48.121 | 35.123 | 2017-03-01 | 01 | 60 
    48.121 | 35.123 | 2017-03-01 | 02 | 15 
    48.121 | 35.123 | 2017-03-01 | 03 | 80 
    48.121 | 35.123 | 2017-03-02 | 01 | 60 
    48.121 | 35.123 | 2017-03-02 | 02 | 15 
    48.121 | 35.123 | 2017-03-02 | 03 | 80 

理论上下面应该发生：

1. 查找数据中的经度和纬度组合的不同值
2. 取上述不同值的数组，并为每个组合创建一个文件，并根据两个参数从源文件中提取数据（长度和012）

我正在努力的是如何开始循环和基于源中的两个参数提取数据，以及如何通过组合的不同值对数据源进行“分区”的参数。

Answer 1

U-SQL不直接支持动态U-SQL，但可以使用“脚本脚本”技术来创建输出。然后，您可以手动运行此输出，或者使用Powershell或Azure Data Factory等软件来运行它。

我已经根据您的测试数据部分基于here的示例创建了一个简单示例。

// Get the initial fileset 
@input = 
    EXTRACT longtitude float, 
      lattitude float, 
      date string, 
      hour int, 
      value1 int, 
      filename string 
    FROM "/input/File_201703{filename}" 
    USING Extractors.Csv(); 


// Add int version of the long and lat columns for grouping on 
@working = 
    SELECT *, 
      (int) longtitude AS int_long, 
      (int) lattitude AS int_lat 
    FROM @input; 


// Work out the filenames 
@filenames = 
    SELECT String.Format("File {0}_{1}.csv", int_long, int_lat) AS outputFilename, 
      int_long, 
      int_lat 
    FROM 
    (
     SELECT int_long, 
       int_lat 
     FROM @working 
     GROUP BY int_long, 
       int_lat 
    ) AS x; 


// Construct the dynamic usql and output it 
@output = 
    SELECT x.xsort, "@input = EXTRACT longtitude float, lattitude float, date string, hour int, value1 int, filename string FROM \"input/File_201703{filename}\" USING Extractors.Csv();" AS usql 
    FROM (VALUES (10)) AS x(xsort) 

    UNION ALL 

    SELECT x.xsort, "@working = SELECT *, (int) longtitude AS int_long, (int) lattitude AS int_lat FROM @input;" AS usql 
    FROM (VALUES (20)) AS x(xsort) 

    UNION ALL 

    SELECT 30 AS xsort, String.Format("OUTPUT (SELECT * FROM @working WHERE int_long == {0} AND int_lat == {1}) TO \"/output/{2}\" USING Outputters.Csv();", int_long, int_lat, outputFilename) AS usql 
    FROM @filenames; 


// Select only the usql column and sort the output 
@output = 
    SELECT usql 
    FROM @output 
ORDER BY xsort 
FETCH 100; 


OUTPUT @output 
TO "/output/dynamic.usql" 
USING Outputters.Text(delimiter : ' ', quoting : false); 

Answer 2

我会建议看文件的概念设定到许多文件（见https://msdn.microsoft.com/en-us/library/azure/mt771650.aspx）和一些建议操作以做基于价值的动态输出，直到该功能可用（见How do I partition a large file into files/directories using only U-SQL and certain fields in the file?作为一个例子）。