基于Azure 数据湖分析与U-SQL实现大数据查询处理

// 脚本：RestaurantScript.usql
// 从CSV文件抽取数据，存储到Rowset变量“employees”中。
@restaurant_ratings =
    EXTRACT rest_id int,
            name string,
            address string,
            online_order bool,
            book_order bool,
            rate double,
            votes int,
            phone string,
            location string,
            rest_type string,
            favorite_dish_id int
    FROM "/Samples/Data/restaurants_ratings.csv"
    USING Extractors.Csv();
// 不执行数据转换：抽取的数据输出到一个TSV文件中。
OUTPUT @restaurant_ratings
TO "/output/restaurants_out.tsv"
USING Outputters.Tsv();

// 脚本：RestaurantScript.usql
// 定义输入、输出文件名和路径的变量。
DECLARE @inputFile = "/Samples/Data/restaurants_ratings.csv";
DECLARE @outputFile = "/output/restaurants_out.tsv";
// 数据从输入CSV文件中抽取，存储在Rowset变量employees中。
@restaurant_ratings =
    EXTRACT rest_id int,
            name string,
            address string,
            online_order bool,
            book_order bool,
            rate double,
            votes int,
            phone string,
            location string,
            rest_type string,
            favorite_dish_id int
    FROM @inputFile
    USING Extractors.Csv(skipFirstNRows:1); // 跳过含有表头的第一行。
// 数据转换操作：重命名数据列，过滤掉行号数据。
@bestOnlineRestaurants =
    SELECT name.ToUpper() AS Name,    // 将餐馆名称转换为大写。
           rate AS Rating,
           online_order AS OnlineOrder,
           phone AS Phone,
           location AS Location,
           rest_type AS Category,
           favorite_dish_id AS FavoriteDishId
    FROM @restaurants_rating
    WHERE rate > 4 && online_order == true;
// 将经转换的数据写入输出文件。
OUTPUT @bestOnlineRestaurants
TO @outputFile
USING Outputters.Tsv(outputHeader:true); // 将列名或表头写入输出文件。

// 相关C#代码文件：RestaurantScript.usql.cs
namespace UsqlApp1
{
    public static class Helpers
    {
        public static string FormatRestaurantName(string name, string location, string restaurantType)
        {
            return name + " (" + restaurantType + ") - " + location;
   // 注意：U-SQL尚不支持新的C# 7.0字符串操作。
               // return $"{name} ( {restaurantType} ) - {location}";
        }
    }
}
// 脚本：RestaurantScript.usql
// 定义输入、输出文件名和路径的变量。
DECLARE @inputFile = "/Samples/Data/restaurants_ratings.csv";
DECLARE @outputFile = "/output/restaurants_out.tsv";
// 数据从输入CSV文件中抽取，存储在Rowset变量employees中。
@restaurant_ratings =
    EXTRACT rest_id int,
            name string,
            address string,
            online_order bool,
            book_order bool,
            rate double,
            votes int,
            phone string,
            location string,
            rest_type string,
            favorite_dish_id int
    FROM @inputFile
    USING Extractors.Csv(skipFirstNRows:1); // 跳过含有表头的第一行。
// 数据转换过程：重命名数据列，滤掉行号。
@bestOnlineRestaurants =
    SELECT USQLApp1.Helpers.FormatRestaurantName(name, location, rest_type) AS Name,
           rate AS Rating,
           online_order AS OnlineOrder,
           phone AS Phone,
           favorite_dish AS FavoriteDish
    FROM @restaurant_ratings
    WHERE rate > 4 && online_order == true;
// 将转换后的数据写入输出文件。
OUTPUT @bestOnlineRestaurants
TO @outputFile
USING Outputters.Tsv(outputHeader:true); // 将列名或表头写入输出文件。

// 脚本：RestaurantScript.usql
DECLARE @inputFile = "/Samples/Data/restaurants_ratings.csv";
DECLARE @outputFile = "/output/restaurants_out.tsv";
// 数据从输入CSV文件中抽取，存储在Rowset变量employees中。
@restaurant_ratings = // 代码同上，此处略。
// 数据转换过程：重命名数据列，过滤掉行号。
@bestOnlineRestaurants = // 代码同上，此处略。

// 定义内存中Rowset变量，存储菜品食材数据，包括菜品标识、菜品名称和食材。
@dish_ingredients =
    SELECT
        * FROM
              (VALUES
               (1, "Biryani", "Rice, Indian spices, Vegetables, Meat, Egg, Yoghurt, Dried Fruits"),
               (2, "Masala Dosa", "rice, husked black gram, mustard seeds, fenugreek seeds, salt, vegetable oil, potatoes, onion, green chillies, curry leaves, turmeric"),
               (3, "Cake", " sugar, butter, egg, cocoa, creme, salt")
              ) AS D(DishId, Dish, Ingredients);
// 在Rowset变量@bestOnlineRestaurants和@dish_ingredients间执行内连接运算。
@rs_innerJn = SELECT
        r.Name,
        r.Rating,
        i.Dish,
        i.Ingredients
FROM @bestOnlineRestaurants AS r
INNER JOIN @dish_ingredients AS i
ON r.FavoriteDishId == i.DishId;
// 结果写入输出文件中。
OUTPUT @rs_innerJn
TO @outputFile
USING Outputters.Tsv(outputHeader:true);

// 定义输入和输出文件的变量。
DECLARE @inputFile = "/Samples/Data/restaurants_raw_data.csv";
DECLARE @outputFile = "/output/restaurants_aggr.csv";
@restaurant_ratings =
    EXTRACT rest_id int,
            name string,
            address string,
            online_order bool,
            book_order bool,
            rate double,
            votes int,
            phone string,
            location string,
            rest_type string,
            favorite_dish_id int
    FROM @inputFile
    USING Extractors.Csv(skipFirstNRows:1);
@output =
    SELECT
           rest_type AS RestaurantType,
           MIN(rate) AS MinRating,
           MAX(rate) AS MaxRating,
           AVG(rate) AS AvgRating,
           STDEV(rate) AS StdDevRating
    FROM @restaurants_ratings
    GROUP BY rest_type;
// 写入到输出文件。
OUTPUT @output
TO @outputFile
USING Outputters.Csv(outputHeader:true);

// 脚本：RestaurantScript.usql
DECLARE @inputFile = "/Samples/Data/restaurants_ratings.csv";
DECLARE @outputFile = "/output/restaurants_out.tsv";
// 数据从输入CSV文件中抽取，存储在Rowset变量employees中。
@restaurant_ratings = // 代码同上，在此略。
// 数据转换过程：过滤出所有评分低于4级的餐馆。
@lowRatedRestaurants =
    SELECT rest_id AS RestaurantId,
           name AS Name,
           rate AS Rating,
           online_order AS OnlineOrder,
           phone AS Phone,
           location AS Location,
           rest_type AS Category,
           favorite_dish_id AS FavoriteDishId
    FROM @restaurants_ratings
    WHERE rate < 4;
// 将低评分餐馆的详细数据插入到U-SQL目录中。
// 判断数据库是否存在。如不存在，则新建数据库。
CREATE DATABASE IF NOT EXISTS RestaurantsDW;
USE RestaurantsDW;
// 判定数据库表是否存在。如已存储，则删除。
DROP TABLE IF EXISTS dbo.LowRatedRestaurants;
// 指定列模式和索引，创建数据库表。
CREATE TABLE dbo.LowRatedRestaurants(
    RestaurantId int,
    Name string,
    INDEX idx
        CLUSTERED (Name DESC)
        DISTRIBUTED BY HASH(Name),
    Rating double,
    OnlineOrder bool,
    Phone string,
    Location string,
    Category string,
    FavoriteDishId int
);
// 将Rowset数据插入到新建的U-SQL数据库表中。
INSERT INTO dbo.LowRatedRestaurants
SELECT * FROM @lowRatedRestaurants;

USE DATABASE RestaurantsDW;
// 判断视图是否存在。如已存在，则删除。
DROP VIEW IF EXISTS RestaurantsView;
// 基于抽取数据创建视图。
CREATE VIEW RestaurantsView AS  
    EXTRACT rest_id int,
            name string,
            address string,
            online_order bool,
            book_order bool,
            rate double,
            votes int,
            phone string,
            location string,
            rest_type string,
            favorite_dish_id int
    FROM "/Samples/Data/restaurants_raw_data.csv"
    USING Extractors.Csv(skipFirstNRows:1); // 跳过包含有表头的第一行

@result = SELECT * FROM RestaurantsDW.dbo.RestaurantsView;
OUTPUT @result
TO "/output/Restaurants_View.csv"
USING Outputters.Csv();

CREATE DATABASE IF NOT EXISTS RestaurantsDW;
USE DATABASE RestaurantsDW;
DROP FUNCTION IF EXISTS tvf_SearchRestaurants;
// 创建TVF，输入参数是字符串类型的餐馆数据，返回包含匹配餐馆详细数据的数据库表。
CREATE FUNCTION tvf_SearchRestaurants(@RestaurantType string)
RETURNS @searchRestaurants TABLE(rest_id int, name string, address string,
                                    online_order bool, book_order bool, rate double,
                                    votes int, phone string, location string,
                                    rest_type string, favorite_dish_id int)
    AS
BEGIN
    @allRestaurants =
        EXTRACT rest_id int,
            name string,
            address string,
            online_order bool,
            book_order bool,
            rate double,
            votes int,
            phone string,
            location string,
            rest_type string,
            favorite_dish_id int
        FROM "/Samples/Data/restaurants_raw_data.csv"
        USING Extractors.Csv(skipFirstNRows:1); // 跳过包含有表头的第一行
    @searchRestaurants =
        SELECT *
        FROM @allRestaurants
        WHERE rest_type == @RestaurantType;
RETURN;
END;

OUTPUT RestaurantsDW.dbo.tvf_SearchRestaurants("Bakery")
TO "/output/BakeryRestaurants.csv"
USING Outputters.Csv();