AWS клей Pyspark, закончить работу с условием? - PullRequest
Новая
       8

AWS клей Pyspark, закончить работу с условием?

0 голосов
/ 03 марта 2020

Похоже, простая задача, но я не могу найти документы, чтобы посмотреть, возможно ли это. По сути, у меня есть склеенная работа, которая выполняется каждый час и ищет папку, чтобы увидеть, были ли загружены данные. В некоторых случаях за последний час не было загружено никаких данных, поэтому, когда функция Glue запускается и видит, что данных нет, я бы хотел, чтобы она прекратилась. Это возможно? Вот некоторый псевдокод для иллюстрации того, что я имею в виду: 

def fn(input):
    *fetches list of data*
    return (list of data)

list_of_data = fn(input)
if list_of_data is None:
    Terminate Job

Ответы [ 3 ]

1 голос
/ 04 марта 2020

Да, поскольку bdcloud правильно упомянул об этом, мы можем напрямую запустить задание Glue из Lambda. Создайте триггер события в целевой папке и, когда файл загружен, запустите задание склеивания. Ниже приведен фрагмент кода для AWS Lambda: 

from __future__ import print_function
import json
import boto3
import time
import sys
import time
from datetime import datetime

s3 = boto3.client('s3')
glue = boto3.client('glue')

def lambda_handler(event, context):
    gluejobname="<< THE GLUE JOB NAME >>"

    try:
        runId = glue.start_job_run(JobName=gluejobname)
        status = glue.get_job_run(JobName=gluejobname, RunId=runId['JobRunId'])
        print("Job Status : ", status['JobRun']['JobRunState'])
    except Exception as e:
        print(e)
        print('Error getting object {} from bucket {}. Make sure they exist '
              'and your bucket is in the same region as this '
              'function.'.format(source_bucket, source_bucket))
    raise e

. Эта настройка работает в производственной среде и успешно работает последние 1,5 года.

Спасибо,

Юв

0 голосов
/ 04 марта 2020

Обозначенный вами psuedocode может работать, так как в прошлом я выполнял подобные задания.

Тем не менее, я обнаружил, что работы с клеем, используемые таким способом, дороги, так как с вас будет взиматься плата за первые 10 минут использования, даже если ваша работа выполнялась менее минуты (особенно в случае отсутствия файлы).

Альтернатива с более высокой стоимостью (но более сложной, поскольку вы будете совместно использовать события S3, SQS и Lambda) заключается в следующем:

  1. Настройка событий уведомление в S3, отслеживающее события PUT в соответствующей папке, которое отправит сообщение в SQS (Simple Queue Service).
  2. Установите период хранения сообщений в очереди SQS на 1 час (или любой другой период, в течение которого вы выполняете задание Glue ). Таким образом, сообщения находятся в очереди не более 1 часа.
  3. Создайте лямбда-задание, которое будет проверять очередь сообщений SQS (используя boto3). По сути, вы положите псевдокод, который у вас есть в лямбде вместо клея. Если есть сообщения (что означает, что по крайней мере 1 файл прибыл за этот период), запустите задание Glue для обработки. Если нет, сделайте что-нибудь и выйдите.

Вышеуказанный метод сэкономит вам $$.

0 голосов
/ 04 марта 2020

Если ваш источник s3, вам даже не нужно запускать задание Glue каждый час, чтобы определить, есть ли какие-либо загрузки / изменения в файлах в исходной корзине s3.

Вы можете использовать лямбда-триггер s3 который фактически запустит вашу работу Glue, если есть какая-либо загрузка на s3. Как только лямбда сработает, вы сможете запустить вашу работу с клеем. Ознакомьтесь с этим видео, чтобы узнать больше.

Таким образом, вы можете запускать свое задание Glue только тогда, когда вместо этого происходит загрузка каждый час.

Если вы все еще хотите запустить свое Склеивать задания каждый час, после чего вы можете использовать Склеивание с закладками , которые обрабатывают только самые последние данные при каждом запуске.

...