Aracılığıyla paylaş

HDInsight .NET SDK'sını kullanarak MapReduce işlerini çalıştırma

  • Makale

HDInsight .NET SDK'sını kullanarak MapReduce işlerini göndermeyi öğrenin. HDInsight kümeleri bazı MapReduce örneklerini içeren bir jar dosyasıyla birlikte gelir. Jar dosyası şeklindedir /example/jars/hadoop-mapreduce-examples.jar. Örneklerden biri wordcount. Wordcount işi göndermek için bir C# konsol uygulaması geliştirirsiniz. İş dosyayı okur /example/data/gutenberg/davinci.txt ve sonuçları öğesine /example/data/davinciwordcountgönderir. Uygulamayı yeniden çalıştırmak istiyorsanız çıkış klasörünü temizlemeniz gerekir.

Not

Bu makaledeki adımlar bir Windows istemcisinden gerçekleştirilmelidir. Hive ile çalışmak için Linux, OS X veya Unix istemcisi kullanma hakkında bilgi için makalenin üst kısmında gösterilen sekme seçiciyi kullanın.

Önkoşullar

HDInsight .NET SDK'sını kullanarak MapReduce işlerini gönderme

HDInsight .NET SDK'sı .NET istemci kitaplıkları sağlar ve bu da .NET'ten HDInsight kümeleriyle çalışmayı kolaylaştırır.

  1. Visual Studio'yu başlatın ve bir C# konsol uygulaması oluşturun.

  2. Araçlar>NuGet Paket Yöneticisi> Paket Yöneticisi Konsolu'na gidin ve aşağıdaki komutu girin:

    Install-Package Microsoft.Azure.Management.HDInsight.Job

  3. Aşağıdaki kodu Program.cs kopyalayın. Ardından, , defaultStorageAccountKeyve defaultStorageContainerNamedeğerlerini existingClusterNameexistingClusterPassworddefaultStorageAccountNameayarlayarak kodu düzenleyin.

    using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading;
using Microsoft.Azure.Management.HDInsight.Job;
using Microsoft.Azure.Management.HDInsight.Job.Models;
using Hyak.Common;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.Storage.Blob;

namespace SubmitHDInsightJobDotNet
{
    class Program
    {
        private static HDInsightJobManagementClient _hdiJobManagementClient;

        private const string existingClusterName = "<Your HDInsight Cluster Name>";
        private const string existingClusterPassword = "<Cluster User Password>";
        private const string defaultStorageAccountName = "<Default Storage Account Name>"; 
        private const string defaultStorageAccountKey = "<Default Storage Account Key>";
        private const string defaultStorageContainerName = "<Default Blob Container Name>";

        private const string existingClusterUsername = "admin";
        private const string existingClusterUri = existingClusterName + ".azurehdinsight.net";
        private const string sourceFile = "/example/data/gutenberg/davinci.txt";
        private const string outputFolder = "/example/data/davinciwordcount";

        static void Main(string[] args)
        {
            System.Console.WriteLine("The application is running ...");

            var clusterCredentials = new BasicAuthenticationCloudCredentials { Username = existingClusterUsername, Password = existingClusterPassword };
            _hdiJobManagementClient = new HDInsightJobManagementClient(existingClusterUri, clusterCredentials);

            SubmitMRJob();

            System.Console.WriteLine("Press ENTER to continue ...");
            System.Console.ReadLine();
        }

        private static void SubmitMRJob()
        {
            List<string> args = new List<string> { { "/example/data/gutenberg/davinci.txt" }, { "/example/data/davinciwordcount" } };

            var paras = new MapReduceJobSubmissionParameters
            {
                JarFile = @"/example/jars/hadoop-mapreduce-examples.jar",
                JarClass = "wordcount",
                Arguments = args
            };

            System.Console.WriteLine("Submitting the MR job to the cluster...");
            var jobResponse = _hdiJobManagementClient.JobManagement.SubmitMapReduceJob(paras);
            var jobId = jobResponse.JobSubmissionJsonResponse.Id;
            System.Console.WriteLine("Response status code is " + jobResponse.StatusCode);
            System.Console.WriteLine("JobId is " + jobId);

            System.Console.WriteLine("Waiting for the job completion ...");

            // Wait for job completion
            var jobDetail = _hdiJobManagementClient.JobManagement.GetJob(jobId).JobDetail;
            while (!jobDetail.Status.JobComplete)
            {
                Thread.Sleep(1000);
                jobDetail = _hdiJobManagementClient.JobManagement.GetJob(jobId).JobDetail;
            }

            // Get job output
            System.Console.WriteLine("Job output is: ");
            var storageAccess = new AzureStorageAccess(defaultStorageAccountName, defaultStorageAccountKey,
                defaultStorageContainerName);

            if (jobDetail.ExitValue == 0)
            {
                // Create the storage account object
                CloudStorageAccount storageAccount = CloudStorageAccount.Parse("DefaultEndpointsProtocol=https;AccountName=" +
                    defaultStorageAccountName +
                    ";AccountKey=" + defaultStorageAccountKey);

                // Create the blob client.
                CloudBlobClient blobClient = storageAccount.CreateCloudBlobClient();

                // Retrieve reference to a previously created container.
                CloudBlobContainer container = blobClient.GetContainerReference(defaultStorageContainerName);

                CloudBlockBlob blockBlob = container.GetBlockBlobReference(outputFolder.Substring(1) + "/part-r-00000");

                using (var stream = blockBlob.OpenRead())
                {
                    using (StreamReader reader = new StreamReader(stream))
                    {
                        while (!reader.EndOfStream)
                        {
                            System.Console.WriteLine(reader.ReadLine());
                        }
                    }
                }
            }
            else
            {
                // fetch stderr output in case of failure
                var output = _hdiJobManagementClient.JobManagement.GetJobErrorLogs(jobId, storageAccess);

                using (var reader = new StreamReader(output, Encoding.UTF8))
                {
                    string value = reader.ReadToEnd();
                    System.Console.WriteLine(value);
                }

            }
        }
    }
}

  4. Uygulamayı çalıştırmak için F5'e basın.

İşi yeniden çalıştırmak için, örnekteki /example/data/davinciwordcountiş çıktı klasörünün adını değiştirmeniz gerekir.

İş başarıyla tamamlandığında, uygulama çıktı dosyasının part-r-00000içeriğini yazdırır.

Sonraki adımlar

Bu makalede, HDInsight kümesi oluşturmanın çeşitli yollarını öğrendiniz. Daha fazla bilgi için aşağıdaki makalelere bakın: