Exécuter des travaux MapReduce avec le Kit de développement logiciel (SDK) .NET HDInsight

  • Article

Apprenez à envoyer des travaux MapReduce à l’aide du Kit de développement logiciel (SDK) .NET HDInsight. Les clusters HDInsight comportent un fichier jar qui contient des exemples MapReduce. Le fichier jar est /example/jars/hadoop-mapreduce-examples.jar. Un des exemples est wordcount. Vous développez une application de console C# pour envoyer une tâche wordcount. Le travail lit le fichier /example/data/gutenberg/davinci.txt et renvoie les résultats vers /example/data/davinciwordcount. Si vous souhaitez réexécuter l’application, vous devez nettoyer le dossier de sortie.

Notes

Les étapes décrites dans cet article doivent être effectuées à partir d'un client Windows. Pour plus d’informations sur l’utilisation d’un client Linux, OS X ou Unix pour utiliser Hive, utilisez le sélecteur d’onglet en haut de l’article.

Prérequis

Envoi de tâches MapReduce avec le Kit de développement logiciel (SDK) .NET HDInsight

Le kit SDK HDInsight .NET fournit des bibliothèques de client .NET, ce qui facilite l’utilisation des clusters HDInsight à partir de .NET.

  1. Démarrez Visual Studio et créez une application console C#.

  2. Accédez à Outils>Gestionnaire de package NuGet>Console du gestionnaire de package, puis entrez la commande suivante :

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

  3. Copiez le code ci-dessous dans Program.cs. Modifiez ensuite le code en définissant les valeurs pour existingClusterName, existingClusterPassword, defaultStorageAccountName, defaultStorageAccountKey et defaultStorageContainerName.

    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. Appuyez sur F5 pour exécuter l'application.

Pour réexécuter le travail, vous devez changer le nom du dossier de sortie du travail. Dans l’exemple, il s’agit de /example/data/davinciwordcount.

Quand le travail se termine correctement, l’application imprime le contenu du fichier de sortie part-r-00000.

Étapes suivantes

Cet article vous a présenté différentes méthodes pour créer un cluster HDInsight. Pour en savoir plus, consultez les articles suivants :