Отправка искрового задания в кластер AWS EMR из aws-cli - PullRequest
Новая
       45

Отправка искрового задания в кластер AWS EMR из aws-cli

0 голосов
/ 04 июня 2018

Я пытаюсь выяснить, как правильно добавить шаг зажигания в мой кластер aws-emr из командной строки aws-cli.

Немного предыстории:

У меня есть большой набор данных (тысячи файлов .csv), который мне нужно прочитать и проанализировать.У меня есть сценарий Python, который выглядит примерно так:

analysis_script.py 

import pandas as pd
from pyspark.sql import SQLContext, DataFrame
from pyspark.sql.types import *
from pyspark import SparkContext
import boto3

#Spark context
sc = SparkContext.getOrCreate()
sqlContext = SQLContext(sc)

df = sqlContext.read.format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat").load("s3n://data_input/*csv")

def analysis(df):
    #do bunch of stuff. Create output dataframe
    return df_output

df_output = analysis(df)

df_output.save_as_csv_to_s3_somehow

Я хочу, чтобы выходной файл CSV перейти в каталог s3: // dataoutput /

Нужно ли добавлять файл py в банку или что-то?Какую команду я использую, чтобы выполнить этот анализ, используя мои узлы кластера, и как я могу получить выходные данные в правильном директое?Спасибо.

Я запускаю кластер, используя: 

aws emr create-cluster --release-label emr-5.5.0\
--name PySpark_Analysis\
--applications Name=Hadoop Name=Hive Name=Spark Name=Pig Name=Ganglia Name=Presto Name=Zeppelin\
--instance-groups InstanceGroupType=MASTER,InstanceCount=1,InstanceType=r3.xlarge InstanceGroupType=CORE,InstanceCount=4,InstanceType=r3.xlarge\
--region us-west-2\
--log-uri s3://emr-logs-zerex/ 
--configurations file://./zeppelin-env-config.json/
--bootstrap-actions Name="Install Python Packages",Path="s3://emr-code/bootstraps/install_python_packages_custom.bash"

1 Ответ

0 голосов
/ 06 июня 2018

Я обычно использую --steps параметр aws emr create-cluster, который можно указать как --steps file://mysteps.json.Файл выглядит следующим образом: 

[
    {
        "Type": "Spark",
        "Name": "KB Spark Program",
        "ActionOnFailure": "TERMINATE_JOB_FLOW",
        "Args": [
            "--verbose",
            "--packages",
            "org.apache.spark:spark-streaming-kafka-0-8_2.11:2.1.1,com.amazonaws:aws-java-sdk-s3:1.11.27,org.apache.hadoop:hadoop-aws:2.7.2,com.databricks:spark-csv_2.11:1.5.0",
            "/tmp/analysis_script.py"
        ]
    },
        {
        "Type": "Spark",
        "Name": "KB Spark Program",
        "ActionOnFailure": "TERMINATE_JOB_FLOW",
        "Args": [
            "--verbose",
            "--packages",
            "org.apache.spark:spark-streaming-kafka-0-8_2.11:2.1.1,com.amazonaws:aws-java-sdk-s3:1.11.27,org.apache.hadoop:hadoop-aws:2.7.2,com.databricks:spark-csv_2.11:1.5.0",
            "/tmp/analysis_script_1.py"
        ]
    }
]

Подробнее о шагах можно прочитать здесь .Я использую скрипт начальной загрузки, чтобы загрузить свой код из S3 в /tmp и затем указать шаги выполнения в файле.

Что касается записи в s3 , здесь - ссылка, объясняющая это.

...