使用 Azure Functions 调用 REST API 并将结果保存在 Azure Data Lake gen2 中

Question

我想调用一个 rest api 并将结果保存为 Azure Data Lake Gen2 中的 csv 或 json 文件。 根据我所阅读的内容，Azure Functions 是可行的方法。

网络服务返回如下格式的数据：

"ID","ProductName","Company"
"1","Apples","Alfreds futterkiste"
"2","Oranges","Alfreds futterkiste"
"3","Bananas","Alfreds futterkiste"
"4","Salad","Alfreds futterkiste"
 ...next rows

我用 C# 编写了一个控制台应用程序，目前将数据输出到控制台。 Web 服务使用分页并返回 1000 行（由 &num 参数确定，最大值为 1000）。在第一个请求之后，我可以使用 &next-parameter 根据 ID 获取接下来的 1000 行。例如网址

http://testWebservice123.com/Example.csv?auth=abc&number=1000&next=1000

将为我提供从 ID 1001 到 2000 的行。 （API 的调用和实际的分页有点复杂，因此我不能使用 Azure Data Factory_v2 来加载到 Azure Data Lake - 这就是我认为我需要 Azure Functions 的原因 - 除非我忽略了另一个服务？？。所以下面只是一个演示，学习如何写入 Azure Data Lake。）

我有以下 C#：

static void Main(string[] args)
    {


        string startUrl = "http://testWebservice123.com/Example.csv?auth=abc&number=1000";
        string url = "";
        string deltaRequestParameter = "";
        string lastLine;
        int numberOfLines = 0;

        do
        {
            url = startUrl + deltaRequestParameter;
            WebClient myWebClient = new WebClient();

            using (Stream myStream = myWebClient.OpenRead(url))
            {

                using (StreamReader sr = new StreamReader(myStream))
                {
                    numberOfLines = 0;
                    while (!sr.EndOfStream)
                    {
                        var row = sr.ReadLine();
                        var values = row.Split(',');

                        //do whatever with the rows by now - i.e. write to console
                        Console.WriteLine(values[0] + " " + values[1]); 

                        lastLine = values[0].Replace("\"", ""); //last line in the loop - get the last ID.
                        numberOfLines++;
                        deltaRequestParameter = "&next=" + lastLine;
                    }

                }

            }
        } while (numberOfLines == 1001); //since the header is returned each time the number of rows will be 1001 until we get to the last request


    }

我想以最有效的方式将数据写入数据湖的 csv 文件。 我将如何重写上述代码以在 Azure Function 中工作并保存到 Azure 数据湖 gen2 中的 csv？

Answer 1

以下是实现结果所需执行的步骤：

1) 创建一个 azure 函数和触发器，您可以保留它 HTTPTrigger/TimerTrigger，或者根据您的需要。

2) 我假设你有代码循环调用 api 直到它给你想要的结果。

3) 将数据存储在内存中后，您必须编写以下代码才能将其写入 Azure 数据湖。

使用您的 c# 代码访问 ADLS 的先决条件：

1) 在 Azure AD 中注册应用

在数据湖存储中授予权限

下面是创建ADLS客户端的代码。

// ADLS connection 
                var adlCreds = GetCreds_SPI_SecretKey(tenantId, ADL_TOKEN_AUDIENCE, serviceAppIDADLS, servicePrincipalSecretADLS);
                var adlsClient = AdlsClient.CreateClient(adlsName, adlCreds);



private static ServiceClientCredentials GetCreds_SPI_SecretKey(string tenant,Uri tokenAudience,string clientId,string secretKey)
        {
            SynchronizationContext.SetSynchronizationContext(new SynchronizationContext());
            var serviceSettings = ActiveDirectoryServiceSettings.Azure;
            serviceSettings.TokenAudience = tokenAudience;
            var creds = ApplicationTokenProvider.LoginSilentAsync(tenant,clientId,secretKey,serviceSettings).GetAwaiter().GetResult();
            return creds;
        }

最后编写实现以将文件保存在 Azure 数据湖中

 const string delim = ",";
        static string adlsInputPath = ConfigurationManager.AppSettings.Get("AdlsInputPath");

public static void ProcessUserProfile(this SampleProfile, AdlsClient adlsClient, string fileNameExtension = "")
        {
            using (MemoryStream memStreamProfile = new MemoryStream())
            {
                using (TextWriter textWriter = new StreamWriter(memStreamProfile))
                {
                    string profile;
                    string header = Helper.GetHeader(delim, Entities.FBEnitities.Profile);
                    string fileName = adlsInputPath + fileNameExtension + "/profile.csv";
                    adlsClient.DataLakeFileHandler(textWriter, header, fileName);
                    profile = socialProfile.UserID                                                
                                    + delim + socialProfile.Profile.First_Name
                                    + delim + socialProfile.Profile.Last_Name
                                    + delim + socialProfile.Profile.Name
                                    + delim + socialProfile.Profile.Age_Range_Min
                                    + delim + socialProfile.Profile.Age_Range_Max
                                    + delim + socialProfile.Profile.Birthday
                                   ;

                    textWriter.WriteLine(profile);
                    textWriter.Flush();
                    memStreamProfile.Flush();
                    adlsClient.DataLakeUpdateHandler(fileName, memStreamProfile);
                }
            }
        }

希望对你有帮助。