AWS Data Pipeline tidak lagi tersedia untuk pelanggan baru. Pelanggan yang sudah ada dari AWS Data Pipeline dapat terus menggunakan layanan seperti biasa. Pelajari selengkapnya
Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
HadoopActivity
Menjalankan MapReduce pekerjaan di cluster. Cluster dapat berupa EMR cluster yang dikelola oleh AWS Data Pipeline atau sumber daya lain jika Anda menggunakannya TaskRunner. Gunakan HadoopActivity saat Anda ingin menjalankan pekerjaan secara paralel. Ini memungkinkan Anda untuk menggunakan sumber penjadwalan YARN kerangka kerja atau negosiator MapReduce sumber daya di Hadoop 1. Jika Anda ingin menjalankan pekerjaan secara berurutan menggunakan tindakan Amazon EMR Step, Anda masih dapat menggunakannya. EmrActivity
Contoh
HadoopActivity menggunakan EMR cluster yang dikelola oleh AWS Data Pipeline
HadoopActivity Objek berikut menggunakan EmrCluster sumber daya untuk menjalankan program:
{ "name": "MyHadoopActivity", "schedule": {"ref": "ResourcePeriod"}, "runsOn": {"ref": “MyEmrCluster”}, "type": "HadoopActivity", "preActivityTaskConfig":{"ref":"preTaskScriptConfig”}, "jarUri": "/home/hadoop/contrib/streaming/hadoop-streaming.jar", "argument": [ "-files", “s3://elasticmapreduce/samples/wordcount/wordSplitter.py“, "-mapper", "wordSplitter.py", "-reducer", "aggregate", "-input", "s3://elasticmapreduce/samples/wordcount/input/", "-output", “s3://test-bucket/MyHadoopActivity/#{@pipelineId}/#{format(@scheduledStartTime,'YYYY-MM-dd')}" ], "maximumRetries": "0", "postActivityTaskConfig":{"ref":"postTaskScriptConfig”}, "hadoopQueue" : “high” }
Berikut ini yang sesuai MyEmrCluster
, yang mengonfigurasi FairScheduler dan mengantri YARN untuk berbasis Hadoop 2: AMIs
{ "id" : "MyEmrCluster", "type" : "EmrCluster", "hadoopSchedulerType" : "PARALLEL_FAIR_SCHEDULING", “amiVersion” : “3.7.0”, "bootstrapAction" : ["s3://
Region
.elasticmapreduce/bootstrap-actions/configure-hadoop,-z,yarn.scheduler.capacity.root.queues=low\,high\,default,-z,yarn.scheduler.capacity.root.high.capacity=50,-z,yarn.scheduler.capacity.root.low.capacity=10,-z,yarn.scheduler.capacity.root.default.capacity=30”] }
Ini adalah yang EmrCluster Anda gunakan untuk mengkonfigurasi FairScheduler di Hadoop 1:
{ "id": "MyEmrCluster", "type": "EmrCluster", "hadoopSchedulerType": "PARALLEL_FAIR_SCHEDULING", "amiVersion": "2.4.8", "bootstrapAction": "s3://
Region
.elasticmapreduce/bootstrap-actions/configure-hadoop,-m,mapred.queue.names=low\\\\,high\\\\,default,-m,mapred.fairscheduler.poolnameproperty=mapred.job.queue.name" }
Berikut ini EmrCluster mengkonfigurasi CapacityScheduler untuk berbasis Hadoop 2: AMIs
{ "id": "MyEmrCluster", "type": "EmrCluster", "hadoopSchedulerType": "PARALLEL_CAPACITY_SCHEDULING", "amiVersion": "3.7.0", "bootstrapAction": "s3://
Region
.elasticmapreduce/bootstrap-actions/configure-hadoop,-z,yarn.scheduler.capacity.root.queues=low\\\\,high,-z,yarn.scheduler.capacity.root.high.capacity=40,-z,yarn.scheduler.capacity.root.low.capacity=60" }
HadoopActivity menggunakan EMR cluster yang sudah ada
Dalam contoh ini, Anda menggunakan workergroups dan a TaskRunner untuk menjalankan program pada cluster yang adaEMR. Definisi pipeline berikut digunakan HadoopActivity untuk:
-
Jalankan MapReduce program hanya pada
myWorkerGroup
sumber daya. Untuk informasi selengkapnya tentang grup pekerja, lihat Menjalankan Pekerjaan pada Sumber Daya yang Ada Menggunakan Runner Tugas. -
Jalankan preActivityTask Config dan Config postActivityTask
{ "objects": [ { "argument": [ "-files", "s3://elasticmapreduce/samples/wordcount/wordSplitter.py", "-mapper", "wordSplitter.py", "-reducer", "aggregate", "-input", "s3://elasticmapreduce/samples/wordcount/input/", "-output", "s3://test-bucket/MyHadoopActivity/#{@pipelineId}/#{format(@scheduledStartTime,'YYYY-MM-dd')}" ], "id": "MyHadoopActivity", "jarUri": "/home/hadoop/contrib/streaming/hadoop-streaming.jar", "name": "MyHadoopActivity", "type": "HadoopActivity" }, { "id": "SchedulePeriod", "startDateTime": "start_datetime", "name": "SchedulePeriod", "period": "1 day", "type": "Schedule", "endDateTime": "end_datetime" }, { "id": "ShellScriptConfig", "scriptUri": "s3://test-bucket/scripts/preTaskScript.sh", "name": "preTaskScriptConfig", "scriptArgument": [ "test", "argument" ], "type": "ShellScriptConfig" }, { "id": "ShellScriptConfig", "scriptUri": "s3://test-bucket/scripts/postTaskScript.sh", "name": "postTaskScriptConfig", "scriptArgument": [ "test", "argument" ], "type": "ShellScriptConfig" }, { "id": "Default", "scheduleType": "cron", "schedule": { "ref": "SchedulePeriod" }, "name": "Default", "pipelineLogUri": "s3://test-bucket/logs/2015-05-22T18:02:00.343Z642f3fe415", "maximumRetries": "0", "workerGroup": "myWorkerGroup", "preActivityTaskConfig": { "ref": "preTaskScriptConfig" }, "postActivityTaskConfig": { "ref": "postTaskScriptConfig" } } ] }
Sintaks
Bidang yang Wajib Diisi | Deskripsi | Jenis Slot |
---|---|---|
jarUri | Lokasi a JAR di Amazon S3 atau sistem file lokal cluster untuk dijalankan. HadoopActivity | String |
Bidang Invokasi Objek | Deskripsi | Jenis Slot |
---|---|---|
jadwal | Objek ini dipanggil dalam pelaksanaan interval jadwal. Pengguna harus menentukan referensi jadwal ke objek lain untuk mengatur urutan eksekusi ketergantungan untuk objek ini. Pengguna dapat memenuhi persyaratan ini dengan secara eksplisit mengatur jadwal pada objek, misalnya, dengan menentukan “jadwal”: {"ref”: "“}. DefaultSchedule Dalam kebanyakan kasus, lebih baik untuk menempatkan referensi jadwal pada objek alur default sehingga semua objek mewarisi jadwal itu. Atau, jika alur memiliki pohon jadwal (jadwal dalam jadwal utama), pengguna dapat membuat objek induk yang memiliki referensi jadwal. Untuk informasi selengkapnya tentang konfigurasi jadwal opsional contoh, lihat https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-schedule.html | Objek Referensi, misalnya “schedule”: {"ref”:” myScheduleId “} |
Grup yang diperlukan (Salah satu dari berikut ini diperlukan) | Deskripsi | Jenis Slot |
---|---|---|
runsOn | EMRCluster tempat pekerjaan ini akan berjalan. | Objek Referensi, misalnya "runsOn“: {" ref”:” myEmrCluster Id "} |
workerGroup | Kelompok pekerja. Ini digunakan untuk tugas perutean. Jika Anda memberikan runsOn nilai dan workerGroup workerGroup ada, diabaikan. | String |
Bidang Opsional | Deskripsi | Jenis Slot |
---|---|---|
argumen | Argumen untuk diteruskan keJAR. | String |
attemptStatus | Baru-baru ini melaporkan status dari aktivitas jarak jauh. | String |
attemptTimeout | Timeout untuk penyelesaian pekerjaan jarak jauh. Jika disetel maka aktivitas jarak jauh yang tidak selesai dalam waktu mulai yang ditetapkan dapat dicoba lagi. | Periode |
dependsOn | Tentukan dependensi pada objek yang bisa dijalankan lainnya. | Objek Referensi, misalnya "dependsOn“: {" ref”:” myActivityId “} |
failureAndRerunModus | Menjelaskan perilaku simpul konsumen ketika dependensi gagal atau menjalankan kembali | Pencacahan |
hadoopQueue | Nama antrean penjadwal Hadoop tempat aktivitas akan dikirimkan. | String |
input | Lokasi data input. | Objek Referensi, misalnya “input”: {"ref”:” myDataNode Id "} |
lateAfterTimeout | Waktu berlalu setelah alur mulai di mana objek harus menyelesaikan. Hal ini dipicu hanya ketika jenis jadwal tidak disetel ke ondemand . |
Periode |
mainClass | Kelas utama dari JAR Anda mengeksekusi dengan HadoopActivity. | String |
maxActiveInstances | Jumlah maksimum instans aktif bersamaan dari suatu komponen. Re-runs tidak dihitung terhadap jumlah instans aktif. | Bilangan Bulat |
maximumRetries | Jumlah maksimum percobaan ulang pada pelanggaran | Bilangan Bulat |
onFail | Tindakan untuk dijalankan ketika objek saat ini gagal. | Objek Referensi, misalnya "onFail“: {" ref”:” myActionId “} |
onLateAction | Tindakan yang harus dipicu jika objek belum dijadwalkan atau masih belum selesai. | Objek Referensi, misalnya "onLateAction“: {" ref”:” myActionId “} |
onSuccess | Tindakan untuk dijalankan ketika objek saat ini berhasil. | Objek Referensi, misalnya "onSuccess“: {" ref”:” myActionId “} |
output | Lokasi data output. | Objek Referensi, misalnya “output”: {"ref”:” myDataNode Id "} |
induk | Induk dari objek saat ini dari mana slot akan diwariskan. | Objek Referensi, misalnya “induk”: {"ref”:” myBaseObject Id "} |
pipelineLogUri | S3 URI (seperti 's3://BucketName/Key/ ') untuk mengunggah log untuk pipeline. | String |
postActivityTaskConfig | Script konfigurasi post-activity yang akan dijalankan. Ini terdiri dari skrip shell di Amazon S3 dan daftar argumen. URI | Objek Referensi, misalnya "postActivityTaskConfig”: {"ref”:” “} myShellScript ConfigId |
preActivityTaskConfig | Script konfigurasi post-activity yang akan dijalankan. Ini terdiri dari skrip shell di Amazon S3 dan daftar argumen. URI | Objek Referensi, misalnya "preActivityTaskConfig”: {"ref”:” “} myShellScript ConfigId |
prasyarat | Mendefinisikan prasyarat secara opsional. Node data tidak ditandai "READY" sampai semua prasyarat terpenuhi. | Objek Referensi, misalnya “prasyarat”: {"ref”:” “myPreconditionId} |
reportProgressTimeout | Batas waktu untuk panggilan berturut-turut pekerjaan jarak jauh ke. reportProgress Jika disetel, maka kegiatan jarak jauh yang tidak melaporkan kemajuan untuk jangka waktu tertentu dapat dianggap terhenti dan jadi dicoba lagi. | Periode |
retryDelay | Durasi timeout antara dua upaya coba lagi. | Periode |
scheduleType | Jenis jadwal mengizinkan Anda untuk menentukan apakah objek dalam definisi alur Anda harus dijadwalkan pada awal interval atau akhir interval. Penjadwalan Gaya Deret Waktu berarti instans dijadwalkan pada akhir setiap interval dan Penjadwalan Gaya Cron berarti intans dijadwalkan pada awal setiap interval. Jadwal sesuai permintaan mengizinkan Anda untuk menjalankan alur satu kali per aktivasi. Ini berarti Anda tidak perlu meng-klon atau membuat ulang alur untuk menjalankannya lagi. Jika Anda menggunakan jadwal sesuai permintaan, itu harus ditentukan dalam objek default dan harus menjadi satu-satunya yang scheduleType ditentukan untuk objek dalam pipeline. Untuk menggunakan saluran pipa sesuai permintaan, Anda cukup memanggil ActivatePipeline operasi untuk setiap proses berikutnya. Nilai adalah: cron, ondemand, dan timeseries. | Pencacahan |
Bidang Runtime | Deskripsi | Jenis Slot |
---|---|---|
@activeInstances | Daftar objek instans aktif terjadwal saat ini. | Objek Referensi, misalnya "activeInstances“: {" ref”:” myRunnableObject Id "} |
@actualEndTime | Waktu ketika eksekusi objek ini selesai. | DateTime |
@actualStartTime | Waktu ketika eksekusi objek ini dimulai. | DateTime |
cancellationReason | cancellationReason Jika objek ini dibatalkan. | String |
@cascadeFailedOn | Deskripsi rantai dependensi tempat objek gagal. | Objek Referensi, misalnya "cascadeFailedOn“: {" ref”:” myRunnableObject Id "} |
emrStepLog | EMRlog langkah hanya tersedia pada upaya EMR aktivitas | String |
errorId | errorId Jika objek ini gagal. | String |
errorMessage | errorMessage Jika objek ini gagal. | String |
errorStackTrace | Jejak tumpukan kesalahan jika objek ini gagal. | String |
@finishedTime | Waktu saat objek ini menyelesaikan eksekusinya. | DateTime |
hadoopJobLog | Log pekerjaan Hadoop tersedia pada upaya untuk aktivitas EMR berbasis. | String |
@healthStatus | Status kondisi objek yang mencerminkan keberhasilan atau kegagalan instans objek terakhir yang mencapai keadaan dihentikan. | String |
@healthStatusFromInstanceId | Id dari objek instans terakhir yang mencapai keadaan dihentikan. | String |
@ healthStatusUpdated Waktu | Waktu di mana status kondisi diperbarui terakhir kali. | DateTime |
hostname | Nama host klien yang mengambil upaya tugas. | String |
@lastDeactivatedTime | Waktu di mana objek ini terakhir dinonaktifkan. | DateTime |
@ latestCompletedRun Waktu | Waktu proses terakhir yang eksekusinya selesai. | DateTime |
@latestRunTime | Waktu proses terakhir untuk eksekusi yang dijadwalkan. | DateTime |
@nextRunTime | Waktu run yang akan dijadwalkan berikutnya. | DateTime |
reportProgressTime | Waktu terbaru bahwa aktivitas jarak jauh melaporkan kemajuan. | DateTime |
@scheduledEndTime | Jadwalkan waktu akhir untuk objek | DateTime |
@scheduledStartTime | Jadwalkan waktu mulai untuk objek | DateTime |
@status | Status objek ini. | String |
@version | Versi alur objek dibuat dengan. | String |
@waitingOn | Deskripsi daftar dependensi objek ini sedang menunggu. | Objek Referensi, misalnya "waitingOn“: {" ref”:” myRunnableObject Id "} |
Bidang Sistem | Deskripsi | Jenis Slot |
---|---|---|
@error | Galat menggambarkan objek yang tidak terbentuk. | String |
@pipelineId | Id dari alur tempat objek ini berada. | String |
@sphere | Lingkup objek menunjukkan tempatnya dalam siklus hidup: Component Objects memunculkan Instance Objects yang mengeksekusi Attempt Objects. | String |