你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站,请访问 https://docs.azure.cn

从数据工厂调用 MapReduce 程序

注意

本文适用于数据工厂版本 1。 如果使用当前版本数据工厂服务,请参阅在数据工厂中使用 MapReduce 活动转换数据

数据工厂管道中的 HDInsight MapReduce 活动会在自己或基于 Windows/Linux 的按需 HDInsight 群集上执行 MapReduce 程序。 本文基于数据转换活动一文,它概述了数据转换和受支持的转换活动。

注意

如果是刚开始接触 Azure 数据工厂,请仔细阅读 Azure 数据工厂简介,并学习教程:生成首个数据管道,然后再阅读本文。

简介

Azure 数据工厂中的管道通过使用链接计算服务来处理链接存储服务中的数据。 它包含一系列活动,其中每个活动执行特定的处理操作。 本文介绍如何使用 HDInsight MapReduce 活动。

请参阅 PigHive,深入了解如何通过使用 HDInsight Pig 和 Hive 活动在管道中基于 Windows/Linux 的 HDInsight 群集上运行 Pig/Hive 脚本。

HDInsight MapReduce 活动的 JSON

在 HDInsight 活动的 JSON 定义中:

  1. 活动类型设置为 HDInsight

  2. className 属性指定类名。

  3. jarFilePath 属性指定 JAR 文件的路径(包括文件名)。

  4. jarLinkedService 属性指定引用包含 JAR 文件的 Azure Blob 存储的链接服务。

  5. arguments 部分中为 MapReduce 程序指定任何参数。 运行时,可在 MapReduce 框架中看到几个额外的参数(例如:mapreduce.job.tags)。 要区分自己的参数和 MapReduce 参数,请考虑将选项和值同时作为参数使用,如下例所示(-s、--input、--output 等选项后紧跟有相应的值)。

    {
        "name": "MahoutMapReduceSamplePipeline",
        "properties": {
            "description": "Sample Pipeline to Run a Mahout Custom Map Reduce Jar. This job calcuates an Item Similarity Matrix to determine the similarity between 2 items",
            "activities": [
                {
                    "type": "HDInsightMapReduce",
                    "typeProperties": {
                        "className": "org.apache.mahout.cf.taste.hadoop.similarity.item.ItemSimilarityJob",
                        "jarFilePath": "adfsamples/Mahout/jars/mahout-examples-0.9.0.2.2.7.1-34.jar",
                        "jarLinkedService": "StorageLinkedService",
                        "arguments": [
                            "-s",
                            "SIMILARITY_LOGLIKELIHOOD",
                            "--input",
                            "wasb://adfsamples@spestore.blob.core.windows.net/Mahout/input",
                            "--output",
                            "wasb://adfsamples@spestore.blob.core.windows.net/Mahout/output/",
                            "--maxSimilaritiesPerItem",
                            "500",
                            "--tempDir",
                            "wasb://adfsamples@spestore.blob.core.windows.net/Mahout/temp/mahout"
                        ]
                    },
                    "inputs": [
                        {
                            "name": "MahoutInput"
                        }
                    ],
                    "outputs": [
                        {
                            "name": "MahoutOutput"
                        }
                    ],
                    "policy": {
                        "timeout": "01:00:00",
                        "concurrency": 1,
                        "retry": 3
                    },
                    "scheduler": {
                        "frequency": "Hour",
                        "interval": 1
                    },
                    "name": "MahoutActivity",
                    "description": "Custom Map Reduce to generate Mahout result",
                    "linkedServiceName": "HDInsightLinkedService"
                }
            ],
            "start": "2017-01-03T00:00:00Z",
            "end": "2017-01-04T00:00:00Z"
        }
    }
    

    可使用 HDInsight MapReduce 活动在 HDInsight 群集中运行任何 MapReduce jar 文件。 在管道的以下示例 JSON 定义中,配置了HDInsight 活动,以便运行 Mahout JAR 文件。

GitHub 上的示例

可从 GitHub 上的数据工厂示例中下载使用 HDInsight MapReduce 活动的示例。

运行字数统计程序

此示例中的管道在 Azure HDInsight 群集上运行字数统计 Map/Reduce 程序。

链接服务

首先,创建一个链接服务,将 Azure HDInsight 群集使用的 Azure 存储链接到 Azure 数据工厂。 如果要复制/粘贴下面的代码,请不要忘记将“帐户名”和“帐户密钥”替换为自己的 Azure 存储 的名称和密钥。

Azure 存储链接服务

{
    "name": "StorageLinkedService",
    "properties": {
        "type": "AzureStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<account name>;AccountKey=<account key>"
        }
    }
}

Azure HDInsight 链接服务

接下来,创建一个链接服务,将 Azure HDInsight 群集链接到 Azure 数据工厂。 如果要复制/粘贴下面的代码,请将 HDInsight 群集名称替换为自己的 HDInsight 群集的名称,并更改用户名和密码值。

{
    "name": "HDInsightLinkedService",
    "properties": {
        "type": "HDInsight",
        "typeProperties": {
            "clusterUri": "https://<HDInsight cluster name>.azurehdinsight.net",
            "userName": "admin",
            "password": "**********",
            "linkedServiceName": "StorageLinkedService"
        }
    }
}

数据集

输出数据集

此示例中的管道不采用任何输入。 为 HDInsight MapReduce 活动指定一个输出数据集。 该数据集仅是推动管道计划所需的一个虚拟数据集。

{
    "name": "MROutput",
    "properties": {
        "type": "AzureBlob",
        "linkedServiceName": "StorageLinkedService",
        "typeProperties": {
            "fileName": "WordCountOutput1.txt",
            "folderPath": "example/data/",
            "format": {
                "type": "TextFormat",
                "columnDelimiter": ","
            }
        },
        "availability": {
            "frequency": "Day",
            "interval": 1
        }
    }
}

管道

此示例中的管道仅具有一个 HDInsightMapReduce 类型的活动。 JSON 中的一些重要属性有:

属性 注释
类型 类型必须设置为 HDInsightMapReduce
className 类名为:wordcount
jarFilePath 包含该类的 jar 文件的路径。 如果要复制/粘贴下面的代码,请不要忘记更改群集的名称。
jarLinkedService 包含 jar 文件的 Azure 存储链接服务。 此链接服务指与 HDInsight 群集关联的存储。
参数 字数统计程序具有两个参数(一个输入和一个输出)。 输入文件是 davinci.txt 文件。
frequency/interval 这些属性的值与输出数据集匹配。
linkedServiceName 指先前创建的 HDInsight 链接服务。
{
    "name": "MRSamplePipeline",
    "properties": {
        "description": "Sample Pipeline to Run the Word Count Program",
        "activities": [
            {
                "type": "HDInsightMapReduce",
                "typeProperties": {
                    "className": "wordcount",
                    "jarFilePath": "<HDInsight cluster name>/example/jars/hadoop-examples.jar",
                    "jarLinkedService": "StorageLinkedService",
                    "arguments": [
                        "/example/data/gutenberg/davinci.txt",
                        "/example/data/WordCountOutput1"
                    ]
                },
                "outputs": [
                    {
                        "name": "MROutput"
                    }
                ],
                "policy": {
                    "timeout": "01:00:00",
                    "concurrency": 1,
                    "retry": 3
                },
                "scheduler": {
                    "frequency": "Day",
                    "interval": 1
                },
                "name": "MRActivity",
                "linkedServiceName": "HDInsightLinkedService"
            }
        ],
        "start": "2014-01-03T00:00:00Z",
        "end": "2014-01-04T00:00:00Z"
    }
}

运行 Spark 程序

可以通过 MapReduce 活动在 HDInsight Spark 群集上运行 Spark 程序。 有关详细信息,请参阅从 Azure 数据工厂调用 Spark 程序

另请参阅