管理 .NET 应用的 Azure Data Lake Analytics

2023-12-20

重要

Azure Data Lake Analytics 于 2024 年 2 月 29 日停用。通过此公告了解更多。

对于数据分析，组织可以使用 Azure Synapse Analytics 或 Microsoft Fabric。

本文介绍如何使用使用 Azure .NET SDK 编写的应用管理 Azure Data Lake Analytics 帐户、数据源、用户和作业。

先决条件

Visual Studio 2015、Visual Studio 2013 更新 4 或安装了 Visual C++ 的 Visual Studio 2012。
适用于 .NET 2.5 或更高版本的 Microsoft Azure SDK。使用 Web 平台安装程序安装它。
所需的 NuGet 包

安装 NuGet 包

封装件	版本
Microsoft.Rest.ClientRuntime.Azure.Authentication	2.3.1
Microsoft.Azure.Management.DataLake.Analytics	3.0.0
Microsoft.Azure.Management.DataLake.Store	2.2.0
Microsoft.Azure.Management.ResourceManager	1.6.0-preview
Microsoft.Azure.Graph.RBAC	3.4.0-preview

可以使用以下命令通过 NuGet 命令行安装这些包：

Install-Package -Id Microsoft.Rest.ClientRuntime.Azure.Authentication  -Version 2.3.1
Install-Package -Id Microsoft.Azure.Management.DataLake.Analytics  -Version 3.0.0
Install-Package -Id Microsoft.Azure.Management.DataLake.Store  -Version 2.2.0
Install-Package -Id Microsoft.Azure.Management.ResourceManager  -Version 1.6.0-preview
Install-Package -Id Microsoft.Azure.Graph.RBAC -Version 3.4.0-preview

常见变量

string subid = "<Subscription ID>"; // Subscription ID (a GUID)
string tenantid = "<Tenant ID>"; // AAD tenant ID or ___domain. For example, "contoso.onmicrosoft.com"
string rg == "<value>"; // Resource  group name
string clientid = "abcdef01-2345-6789-0abc-def012345678"; // Sample client ID

身份验证

可以使用多个选项登录到 Azure Data Lake Analytics。以下代码片段演示了一个示例，其中显示了使用弹出窗口进行交互式用户身份验证的身份验证示例。

对于 ClientID，可以使用用户的 ID，也可以使用服务主体的应用程序（客户端）ID。

using System;
using System.IO;
using System.Threading;
using System.Security.Cryptography.X509Certificates;

using Microsoft.Rest;
using Microsoft.Rest.Azure.Authentication;
using Microsoft.Azure.Management.DataLake.Analytics;
using Microsoft.Azure.Management.DataLake.Analytics.Models;
using Microsoft.Azure.Management.DataLake.Store;
using Microsoft.Azure.Management.DataLake.Store.Models;
using Microsoft.IdentityModel.Clients.ActiveDirectory;
using Microsoft.Azure.Graph.RBAC;

public static Program
{
   public static string TENANT = "microsoft.onmicrosoft.com";
   public static string CLIENTID = "abcdef01-2345-6789-0abc-def012345678";
   public static System.Uri ARM_TOKEN_AUDIENCE = new System.Uri( @"https://management.core.windows.net/");
   public static System.Uri ADL_TOKEN_AUDIENCE = new System.Uri( @"https://datalake.azure.net/" );
   public static System.Uri GRAPH_TOKEN_AUDIENCE = new System.Uri( @"https://graph.windows.net/" );

   static void Main(string[] args)
   {
      string MY_DOCUMENTS= System.Environment.GetFolderPath( System.Environment.SpecialFolder.MyDocuments);
      string TOKEN_CACHE_PATH = System.IO.Path.Combine(MY_DOCUMENTS, "my.tokencache");

      var tokenCache = GetTokenCache(TOKEN_CACHE_PATH);
      var armCreds = GetCreds_User_Popup(TENANT, ARM_TOKEN_AUDIENCE, CLIENTID, tokenCache);
      var adlCreds = GetCreds_User_Popup(TENANT, ADL_TOKEN_AUDIENCE, CLIENTID, tokenCache);
      var graphCreds = GetCreds_User_Popup(TENANT, GRAPH_TOKEN_AUDIENCE, CLIENTID, tokenCache);
   }
}

GetCreds_User_Popup 的源代码以及用于身份验证的其他选项的代码涵盖于 Data Lake Analytics .NET 身份验证选项中。

创建客户端管理对象

var resourceManagementClient = new ResourceManagementClient(armCreds) { SubscriptionId = subid };

var adlaAccountClient = new DataLakeAnalyticsAccountManagementClient(armCreds);
adlaAccountClient.SubscriptionId = subid;

var adlsAccountClient = new DataLakeStoreAccountManagementClient(armCreds);
adlsAccountClient.SubscriptionId = subid;

var adlaCatalogClient = new DataLakeAnalyticsCatalogManagementClient(adlCreds);
var adlaJobClient = new DataLakeAnalyticsJobManagementClient(adlCreds);

var adlsFileSystemClient = new DataLakeStoreFileSystemManagementClient(adlCreds);

var  graphClient = new GraphRbacManagementClient(graphCreds);
graphClient.TenantID = ___domain;

管理账户

创建 Azure 资源组

如果尚未创建一个资源组，则必须有一个 Azure 资源组才能创建 Data Lake Analytics 组件。需要身份验证凭据、订阅 ID 和位置。以下代码演示如何创建资源组：

var resourceGroup = new ResourceGroup { Location = ___location };
resourceManagementClient.ResourceGroups.CreateOrUpdate(groupName, rg);

有关详细信息，请参阅 Azure 资源组和 Data Lake Analytics。

创建 Data Lake Store 帐户

Ever ADLA 帐户需要 ADLS 帐户。如果还没有要使用的代码，可以使用以下代码创建一个：

var new_adls_params = new DataLakeStoreAccount(___location: _location);
adlsAccountClient.Account.Create(rg, adls, new_adls_params);

创建 Data Lake Analytics 帐户

以下代码创建 ADLS 帐户

var new_adla_params = new DataLakeAnalyticsAccount()
{
   DefaultDataLakeStoreAccount = adls,
   Location = ___location
};

adlaClient.Account.Create(rg, adla, new_adla_params);

列出 Data Lake Store 账户

var adlsAccounts = adlsAccountClient.Account.List().ToList();
foreach (var adls in adlsAccounts)
{
   Console.WriteLine($"ADLS: {0}", adls.Name);
}

列出 Data Lake Analytics 帐户

var adlaAccounts = adlaClient.Account.List().ToList();

for (var adla in AdlaAccounts)
{
   Console.WriteLine($"ADLA: {0}, adla.Name");
}

检查帐户是否存在

bool exists = adlaClient.Account.Exists(rg, adla));

获取有关帐户的信息

bool exists = adlaClient.Account.Exists(rg, adla));
if (exists)
{
   var adla_accnt = adlaClient.Account.Get(rg, adla);
}

删除帐户

if (adlaClient.Account.Exists(rg, adla))
{
   adlaClient.Account.Delete(rg, adla);
}

获取默认 Data Lake Store 帐户

每个 Data Lake Analytics 帐户都需要默认的 Data Lake Store 帐户。使用此代码确定分析帐户的默认应用商店帐户。

if (adlaClient.Account.Exists(rg, adla))
{
  var adla_accnt = adlaClient.Account.Get(rg, adla);
  string def_adls_account = adla_accnt.DefaultDataLakeStoreAccount;
}

管理数据源

Data Lake Analytics 目前支持以下数据源：

连接到 Azure 存储帐户

可以创建指向 Azure 存储帐户的链接。

string storage_key = "xxxxxxxxxxxxxxxxxxxx";
string storage_account = "mystorageaccount";
var addParams = new AddStorageAccountParameters(storage_key);            
adlaClient.StorageAccounts.Add(rg, adla, storage_account, addParams);

列出 Azure 存储数据源

var stg_accounts = adlaAccountClient.StorageAccounts.ListByAccount(rg, adla);

if (stg_accounts != null)
{
  foreach (var stg_account in stg_accounts)
  {
      Console.WriteLine($"Storage account: {0}", stg_account.Name);
  }
}

列出 Data Lake Store 数据源

var adls_accounts = adlsClient.Account.List();

if (adls_accounts != null)
{
  foreach (var adls_accnt in adls_accounts)
  {
      Console.WriteLine($"ADLS account: {0}", adls_accnt.Name);
  }
}

上传和下载文件夹和文件

可以使用 Data Lake Store 文件系统客户端管理对象，使用以下方法将单个文件或文件夹从 Azure 上传到本地计算机：

上传文件夹
上传文件
下载文件夹
下载文件

这些方法的第一个参数是 Data Lake Store 帐户的名称，后跟源路径和目标路径的参数。

以下示例演示如何在 Data Lake Store 中下载文件夹。

adlsFileSystemClient.FileSystem.DownloadFolder(adls, sourcePath, destinationPath);

在 Data Lake Store 帐户中创建文件

using (var memstream = new MemoryStream())
{
   using (var sw = new StreamWriter(memstream, UTF8Encoding.UTF8))
   {
      sw.WriteLine("Hello World");
      sw.Flush();
      
      memstream.Position = 0;

      adlsFileSystemClient.FileSystem.Create(adls, "/Samples/Output/randombytes.csv", memstream);
   }
}

验证 Azure 存储帐户路径

以下代码检查 Data Lake Analytics 帐户（analyticsAccountName）中是否存在 Azure 存储帐户（storageAccntName），以及 Azure 存储帐户中是否存在容器（containerName）。

string storage_account = "mystorageaccount";
string storage_container = "mycontainer";
bool accountExists = adlaClient.Account.StorageAccountExists(rg, adla, storage_account));
bool containerExists = adlaClient.Account.StorageContainerExists(rg, adla, storage_account, storage_container));

管理目录和作业

DataLakeAnalyticsCatalogManagementClient 对象提供了管理为每个 Azure Data Lake Analytics 帐户提供的 SQL 数据库的方法。 DataLakeAnalyticsJobManagementClient 提供了使用 U-SQL 脚本在数据库上运行的提交和管理作业的方法。

列出数据库和架构

在可以列出的几个方面中，最常见的是数据库及其架构。以下代码获取数据库集合，然后枚举每个数据库的架构。

var databases = adlaCatalogClient.Catalog.ListDatabases(adla);
foreach (var db in databases)
{
  Console.WriteLine($"Database: {db.Name}");
  Console.WriteLine(" - Schemas:");
  var schemas = adlaCatalogClient.Catalog.ListSchemas(adla, db.Name);
  foreach (var schm in schemas)
  {
      Console.WriteLine($"\t{schm.Name}");
  }
}

列出数据表的列

以下代码演示如何使用 Data Lake Analytics 目录管理客户端访问数据库，以列出指定表中的列。

var tbl = adlaCatalogClient.Catalog.GetTable(adla, "master", "dbo", "MyTableName");
IEnumerable<USqlTableColumn> columns = tbl.ColumnList;

foreach (USqlTableColumn utc in columns)
{
  Console.WriteLine($"\t{utc.Name}");
}

提交 U-SQL 作业

以下代码演示如何使用 Data Lake Analytics 作业管理客户端提交作业。

string scriptPath = "/Samples/Scripts/SearchResults_Wikipedia_Script.txt";
Stream scriptStrm = adlsFileSystemClient.FileSystem.Open(_adlsAccountName, scriptPath);
string scriptTxt = string.Empty;
using (StreamReader sr = new StreamReader(scriptStrm))
{
    scriptTxt = sr.ReadToEnd();
}

var jobName = "SR_Wikipedia";
var jobId = Guid.NewGuid();
var properties = new USqlJobProperties(scriptTxt);
var parameters = new JobInformation(jobName, JobType.USql, properties, priority: 1, degreeOfParallelism: 1, jobId: jobId);
var jobInfo = adlaJobClient.Job.Create(adla, jobId, parameters);
Console.WriteLine($"Job {jobName} submitted.");

列出失败的作业

以下代码列出了有关失败的作业的信息。

var odq = new ODataQuery<JobInformation> { Filter = "result eq 'Failed'" };
var jobs = adlaJobClient.Job.List(adla, odq);
foreach (var j in jobs)
{
   Console.WriteLine($"{j.Name}\t{j.JobId}\t{j.Type}\t{j.StartTime}\t{j.EndTime}");
}

列出管道

以下代码列出了有关提交到帐户的每个作业流程的信息。

var pipelines = adlaJobClient.Pipeline.List(adla);
foreach (var p in pipelines)
{
   Console.WriteLine($"Pipeline: {p.Name}\t{p.PipelineId}\t{p.LastSubmitTime}");
}

列出重复事件

以下代码列出了有关提交到帐户的每个作业重复周期的信息。

var recurrences = adlaJobClient.Recurrence.List(adla);
foreach (var r in recurrences)
{
   Console.WriteLine($"Recurrence: {r.Name}\t{r.RecurrenceId}\t{r.LastSubmitTime}");
}

常见图表场景

在 Microsoft Entra ID 目录中查找用户

var userinfo = graphClient.Users.Get( "bill@contoso.com" );

获取 Microsoft Entra ID 目录中用户的 ObjectId

var userinfo = graphClient.Users.Get( "bill@contoso.com" );
Console.WriteLine( userinfo.ObjectId )

管理计算策略

DataLakeAnalyticsAccountManagementClient 对象提供了管理 Data Lake Analytics 帐户计算策略的方法。

列出计算策略

以下代码检索 Data Lake Analytics 帐户的计算策略列表。

var policies = adlaAccountClient.ComputePolicies.ListByAccount(rg, adla);
foreach (var p in policies)
{
   Console.WriteLine($"Name: {p.Name}\tType: {p.ObjectType}\tMax AUs / job: {p.MaxDegreeOfParallelismPerJob}\tMin priority / job: {p.MinPriorityPerJob}");
}

创建新的计算策略

以下代码为 Data Lake Analytics 帐户创建新的计算策略，将指定用户可用的最大 AU 设置为 50，并将最小作业优先级设置为 250。

var userAadObjectId = "3b097601-4912-4d41-b9d2-78672fc2acde";
var newPolicyParams = new ComputePolicyCreateOrUpdateParameters(userAadObjectId, "User", 50, 250);
adlaAccountClient.ComputePolicies.CreateOrUpdate(rg, adla, "GaryMcDaniel", newPolicyParams);

后续步骤

Microsoft Azure 数据湖分析概述
使用 Azure 门户管理 Azure Data Lake Analytics
使用 Azure 门户监视和排查 Azure Data Lake Analytics 作业问题

通过