PigActivity - AWS Data Pipeline
ContohSintaksisLihat Juga

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

PigActivity

PigActivitymenyediakan dukungan asli untuk skrip Pig diAWS Data Pipelinetanpa persyaratan untuk menggunakanShellCommandActivityatauEmrActivity. Selain itu,PigActivitymendukung pementasan data. Ketika bidang stage diatur ke BETUL, AWS Data Pipeline men-stage data input sebagai skema di Pig tanpa kode tambahan dari pengguna.

Contoh

Contoh alur berikut menunjukkan cara menggunakan PigActivity. Contoh alur melakukan langkah-langkah berikut:

  • MyPigActivity1 memuat data dari Amazon S3 dan menjalankan skrip Pig yang memilih beberapa kolom data dan mengunggahnya ke Amazon S3.

  • MyPigActivity2 memuat output pertama, memilih beberapa kolom dan tiga baris data, dan mengunggahnya ke Amazon S3 sebagai output kedua.

  • MyPigActivity3 memuat data keluaran kedua, menyisipkan dua baris data dan hanya kolom bernama “kelima” ke Amazon RDS.

  • MyPigActivity4 memuat data Amazon RDS, memilih baris pertama data, dan mengunggahnya ke Amazon S3.

{
  "objects": [
    {
      "id": "MyInputData1",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "directoryPath": "s3://example-bucket/pigTestInput",
      "name": "MyInputData1",
      "dataFormat": {
        "ref": "MyInputDataType1"
      },
      "type": "S3DataNode"
    },
    {
      "id": "MyPigActivity4",
      "scheduleType": "CRON",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "input": {
        "ref": "MyOutputData3"
      },
      "pipelineLogUri": "s3://example-bucket/path/",
      "name": "MyPigActivity4",
      "runsOn": {
        "ref": "MyEmrResource"
      },
      "type": "PigActivity",
      "dependsOn": {
        "ref": "MyPigActivity3"
      },
      "output": {
        "ref": "MyOutputData4"
      },
      "script": "B = LIMIT ${input1} 1; ${output1} = FOREACH B GENERATE one;",
      "stage": "true"
    },
    {
      "id": "MyPigActivity3",
      "scheduleType": "CRON",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "input": {
        "ref": "MyOutputData2"
      },
      "pipelineLogUri": "s3://example-bucket/path",
      "name": "MyPigActivity3",
      "runsOn": {
        "ref": "MyEmrResource"
      },
      "script": "B = LIMIT ${input1} 2; ${output1} = FOREACH B GENERATE Fifth;",
      "type": "PigActivity",
      "dependsOn": {
        "ref": "MyPigActivity2"
      },
      "output": {
        "ref": "MyOutputData3"
      },
      "stage": "true"
    },
    {
      "id": "MyOutputData2",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "name": "MyOutputData2",
      "directoryPath": "s3://example-bucket/PigActivityOutput2",
      "dataFormat": {
        "ref": "MyOutputDataType2"
      },
      "type": "S3DataNode"
    },
    {
      "id": "MyOutputData1",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "name": "MyOutputData1",
      "directoryPath": "s3://example-bucket/PigActivityOutput1",
      "dataFormat": {
        "ref": "MyOutputDataType1"
      },
      "type": "S3DataNode"
    },
    {
      "id": "MyInputDataType1",
      "name": "MyInputDataType1",
      "column": [
        "First STRING",
        "Second STRING",
        "Third STRING",
        "Fourth STRING",
        "Fifth STRING",
        "Sixth STRING",
        "Seventh STRING",
        "Eighth STRING",
        "Ninth STRING",
        "Tenth STRING"
      ],
      "inputRegEx": "^(\\\\S+) (\\\\S+) (\\\\S+) (\\\\S+) (\\\\S+) (\\\\S+) (\\\\S+) (\\\\S+) (\\\\S+) (\\\\S+)",
      "type": "RegEx"
    },
    {
      "id": "MyEmrResource",
      "region": "us-east-1",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "keyPair": "example-keypair",
      "masterInstanceType": "m1.small",
      "enableDebugging": "true",
      "name": "MyEmrResource",
      "actionOnTaskFailure": "continue",
      "type": "EmrCluster"
    },
    {
      "id": "MyOutputDataType4",
      "name": "MyOutputDataType4",
      "column": "one STRING",
      "type": "CSV"
    },
    {
      "id": "MyOutputData4",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "directoryPath": "s3://example-bucket/PigActivityOutput3",
      "name": "MyOutputData4",
      "dataFormat": {
        "ref": "MyOutputDataType4"
      },
      "type": "S3DataNode"
    },
    {
      "id": "MyOutputDataType1",
      "name": "MyOutputDataType1",
      "column": [
        "First STRING",
        "Second STRING",
        "Third STRING",
        "Fourth STRING",
        "Fifth STRING",
        "Sixth STRING",
        "Seventh STRING",
        "Eighth STRING"
      ],
      "columnSeparator": "*",
      "type": "Custom"
    },
    {
      "id": "MyOutputData3",
      "username": "___",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "insertQuery": "insert into #{table} (one) values (?)",
      "name": "MyOutputData3",
      "*password": "___",
      "runsOn": {
        "ref": "MyEmrResource"
      },
      "connectionString": "jdbc:mysql://example-database-instance:3306/example-database",
      "selectQuery": "select * from #{table}",
      "table": "example-table-name",
      "type": "MySqlDataNode"
    },
    {
      "id": "MyOutputDataType2",
      "name": "MyOutputDataType2",
      "column": [
        "Third STRING",
        "Fourth STRING",
        "Fifth STRING",
        "Sixth STRING",
        "Seventh STRING",
        "Eighth STRING"
      ],
      "type": "TSV"
    },
    {
      "id": "MyPigActivity2",
      "scheduleType": "CRON",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "input": {
        "ref": "MyOutputData1"
      },
      "pipelineLogUri": "s3://example-bucket/path",
      "name": "MyPigActivity2",
      "runsOn": {
        "ref": "MyEmrResource"
      },
      "dependsOn": {
        "ref": "MyPigActivity1"
      },
      "type": "PigActivity",
      "script": "B = LIMIT ${input1} 3; ${output1} = FOREACH B GENERATE Third, Fourth, Fifth, Sixth, Seventh, Eighth;",
      "output": {
        "ref": "MyOutputData2"
      },
      "stage": "true"
    },
    {
      "id": "MyEmrResourcePeriod",
      "startDateTime": "2013-05-20T00:00:00",
      "name": "MyEmrResourcePeriod",
      "period": "1 day",
      "type": "Schedule",
      "endDateTime": "2013-05-21T00:00:00"
    },
    {
      "id": "MyPigActivity1",
      "scheduleType": "CRON",
      "schedule": {
        "ref": "MyEmrResourcePeriod"
      },
      "input": {
        "ref": "MyInputData1"
      },
      "pipelineLogUri": "s3://example-bucket/path",
      "scriptUri": "s3://example-bucket/script/pigTestScipt.q",
      "name": "MyPigActivity1",
      "runsOn": {
        "ref": "MyEmrResource"
      },
      "scriptVariable": [
        "column1=First",
        "column2=Second",
        "three=3"
      ],
      "type": "PigActivity",
      "output": {
        "ref": "MyOutputData1"
      },
      "stage": "true"
    }
  ]
}

Isi dari pigTestScript.q adalah sebagai berikut.

B = LIMIT ${input1} $three; ${output1} = FOREACH B GENERATE $column1, $column2, Third, Fourth, Fifth, Sixth, Seventh, Eighth;

Sintaksis

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 menetapkan 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, “jadwal”: {"ref”:”myScheduleId“}

Grup yang diperlukan (Salah satu dari berikut ini diperlukan) Deskripsi Jenis Slot
script Script Pig yang akan dijalankan. String
scriptUri Lokasi script Pig untuk menjalankan (misalnya, s3://scriptLocation). String

Grup yang diperlukan (Salah satu dari berikut ini diperlukan) Deskripsi Jenis Slot
runsOn ESDM Cluster di mana iniPigActivityberjalan. Objek Referensi, misalnya, “RunsOn”: {"ref”:”myEmrClusterId "}
workerGroup Kelompok pekerja. Ini digunakan untuk tugas perutean. Jika Anda memberikan nilai runsOn dan workerGroup ada, workerGroup akan diabaikan. String

Bidang Opsional Deskripsi Jenis Slot
attemptStatus Status yang paling baru dilaporkan dari aktivitas jarak jauh. String
attemptTimeout Timeout untuk penyelesaian pekerjaan jarak jauh. Jika disetel, maka aktivitas jarak jauh yang tidak lengkap dalam waktu mulai yang ditetapkan mungkin dicoba lagi. Periode
dependsOn Menentukan 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
input Sumber data input. Objek Referensi, misalnya, “input”: {"ref”:”myDataNodeId "}
lateAfterTimeout Waktu berlalu setelah alur mulai di mana objek harus menyelesaikan. Hal ini dipicu hanya ketika jenis jadwal tidak disetel ke ondemand. Periode
maxActiveInstances Jumlah maksimum instans aktif bersamaan dari suatu komponen. Re-runs tidak dihitung terhadap jumlah instans aktif. Bulat
maximumRetries Upaya jumlah maksimum mencoba lagi pada kegagalan. 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 Sumber data output. Objek Referensi, misalnya, “output”: {"ref”:”myDataNodeId "}
induk Induk dari objek saat ini dari mana slot akan diwariskan. Objek Referensi, misalnya, “parent”: {"ref”:”myBaseObjectId "}
pipelineLogUri URI Amazon S3 (seperti 's3://BucketName/Key/ ') untuk mengunggah log untuk pipeline. String
postActivityTaskKonfigurasi Script konfigurasi post-activity yang akan dijalankan. Ini terdiri dari URI script shell di Amazon S33 dan daftar argumen. Objek Referensi, misalnya,”postActivityTaskKonfigurasi”: {"ref”:”myShellScriptConfigId“}
preActivityTaskKonfigurasi Script konfigurasi post-activity yang akan dijalankan. Ini terdiri dari URI shell script di Amazon S3 dan daftar argumen. Objek Referensi, misalnya,”preActivityTaskKonfigurasi”: {"ref”:”myShellScriptConfigId“}
prasyarat Mendefinisikan prasyarat secara opsional. Sebuah simpul data tidak ditandai "READY" sampai semua prasyarat telah terpenuhi. Objek Referensi, misalnya, “prasyarat”: {"ref”:”myPreconditionId“}
reportProgressTimeout Timeout untuk panggilan kerja jarak jauh berturut-turut ke reportProgress. Jika disetel, maka kegiatan jarak jauh yang tidak melaporkan kemajuan untuk jangka waktu tertentu dapat dianggap terhenti dan jadi dicoba lagi. Periode
resizeClusterBeforeMenjalankan Mengubah ukuran klaster sebelum melakukan aktivitas ini untuk mengakomodasi simpul data DynamoDB ditentukan sebagai input atau output.
catatan

Jika aktivitas Anda menggunakan DynamoDBDataNode baik sebagai input atau output simpul data, dan jika Anda menyetel resizeClusterBeforeRunning ke TRUE, AWS Data Pipeline mulai menggunakan tipe instans m3.xlarge. Ini akan menimpa pilihan tipe instans Anda dengan m3.xlarge, yang dapat menambah biaya bulanan Anda.

 Boolean
resizeClusterMaxContoh Batas pada jumlah maksimum instans yang dapat diminta oleh algoritme resize. Bulat
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 scheduleType yang ditentukan untuk objek dalam alur. Untuk menggunakan jaringan pipa sesuai permintaan, Anda cukup menghubungiActivatePipelineoperasi untuk setiap run berikutnya. Nilai adalah: cron, ondemand, dan timeseries. Pencacahan
scriptVariable Argumen untuk diteruskan ke script Pig. Anda dapat menggunakan scriptVariable dengan script atau scriptUri. String
stage Menentukan apakah staging diaktifkan dan mengizinkan script Pig Anda untuk memiliki akses ke tabel data ter-staged, seperti ${INPUT1} dan ${OUTPUT1}. Boolean

Bidang Runtime Deskripsi Jenis Slot
@activeInstances Daftar objek instans aktif terjadwal saat ini. Referensi Objek, misalnya, “ActiveInstances”: {"ref”:”myRunnableObjectId "}
@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”:”myRunnableObjectId "}
emrStepLog Langkah log Amazon EMR tersedia hanya pada upaya aktivitas EMR. 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 tugas Hadoop tersedia pada upaya untuk kegiatan berbasis EMR. 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
@healthStatusUpdatedWaktu 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
@latestCompletedRunWaktu 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 akhir waktu untuk objek. DateTime
@scheduledStartTime Jadwalkan waktu mulai untuk objek. DateTime
@status Status objek ini. String
@version Versi alur tempat objek dibuat. String
@waitingOn Deskripsi daftar dependensi objek ini sedang menunggu. Objek Referensi, misalnya, “WaitingOn”: {"ref”:”myRunnableObjectId "}

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

Lihat Juga