How to use EMR to process lot of logs?

In the previous article, we looked into the particular Big data example and tried to handle it with streaming. As a result, a thought crossed our minds that for this case streaming can not be used to meet all of the requirements.

 

First, let us imagine that you have a series of request logs files, each line records 1 request.

Log files are delivered out of order.

Some are delivered within 60 seconds of the requests, while other log files are delivered hours after the requests happen.

There are 10k files per hour each with ~10000 rows of requests.

 

2017-09-01T12:00:01,/v1/boxes/box.json

2017-09-01T12:00:02,/v1/boxes/box1.json

2017-09-01T12:00:02,/v1/boxes/box2.json

2017-09-01T12:00:03,/v1/boxes/box3.json

2017-09-01T12:00:02,/v1/boxes/box4.json

 

* Log file example, that includes timestamp, and uri of request. We want to keep a count of the the number of requests per unique uri per hour.

* It’s important that this count is available as soon after the requests happen as possible, even if incomplete.

* The count should update as delayed log files are delivered.

* The process should expect failures, reprocessing data shouldn’t result in over counting.

We need to provide an explanation for how we would:

* Process log files

* Handle delayed data

* Reprocess log files without over counting requests

If there are any cases where system failure would result in incorrect counts. Continue to handle growing scale of this data set over time.

This can take the form of which concepts you would apply and how, which technology you’d use, how and why you’d pick those technologies, pseudo code, diagrams, or any other means to express your solution.

How quickly do you think this ETL might take? What happens if your data ingestion is 10x bigger? 100x bigger?

What parts of your infrastructure will be stressed the most?

 

Let us consider Lambda AWS service as another Data Engineer advised. Our assumption is to parse the data with Lambda and then to load it into database. In this case, we need somehow re-count the files and there can be lot of pitfalls.

How can it be designed? We can apply trigger event on coming files. Then Lambda besides processing can increment counter in database or even insert file, data, status.

After that, we can run Glue job to re-process data if we have the same filename as it was. Glue is better than lambda for this case because It is run on EMR under the hood (but this cluster can not be so flexibly configured as emr). Actually, lambda will not be able to process really big amount of data in this case.

One Lambda function has only one or two cores available, however, the power in Lambda is that you can run hundreds of them at the same time. The default limit of the amount of active Lambda functions is 100, but this is just a limit to safeguard the infrastructure (and your wallet). You can ask for more.

So your account can have 100 Lambda functions running at the same time, which you could see as 100 cores (however it's not). If you request a limit increase, this could also be 1000 or 10,000 or 100,000.

Analogy: instead of having 1 100-core computer, you have 100 1-core computers. While AWS till today didn't disclose any details regarding processing power of AWS Lambda functions, they now did after announcing an increase of the maximum memory from 1536 MB to 3008 MB.

But even such amount memory is not enough to process the data came even for 1 hour. Therefore, the way of data processing for this case is either Glue or EMR.

I would definitely prefer EMR despite of the fact that it requires more skills and expertise rather than Glue.

1) 1 hour tells us that it is a typical situation for batch processing. Therefore, we can create a Spark workflow to process it (EMR cluster)

2) We can ingest the data to some staging table (staging_logs). There would be filename, jsonurl, timestamp, date_hour_partition (we can retrieve it from timestamp column). We can use a such partition structure for logs year=YYYY/month=MM/day=dd/hour=HH

3) Then we need to create another table (dwh_logs) in data_warehouse database and load only the newest data there

To do it we can just :

## choose only the newest data. If it is possible we can also clarify the requirements, maybe we can decrease the period of possible reprocessing with for example 1 latest month 

Rule of thumb: use as less data as it is possible. It is even better to reduce it to 1 day. "Some are delivered within 60 seconds of the requests, while other log files are delivered hours after the requests happen." it means that files can be delayed not for days but for hours. So following the requirement, we can assume that all of the delayed data should come in the frame of 24 hours.

hive_ctx.sql("""

     select

             filename, year, month,

            day, hour

    from

           (select distinct filename, year, month, day, hour

           from {staging}.staging_logs

           where year = year(current_date)

               and month >= month(current_date)-1) res1

except

          (select distinct filename, year, month, day, hour

          from {data_warehouse}.dwh_logs

          where year = year(current_date)

             and month >= month(current_date)-1) res2

""".format(...)

).repartition('year', 'month', 'day', hour').

createOrReplaceTempView('to_load_files')

 

## apply count function only to the filtered data. Load only delta to dwh table

sql = """

with res

         as (

                select stlogs.filename,

                            stlogs.year, stlogs.month, stlogs.day, stlogs.hour ,

                           stlogs.jsonurl,

                          stlogs.timestamp

               from {staging}.staging_logs stlogs

                 join to_load_files lf on stlogs.year = lf.year

                    and stlogs.month = lf.month,

                    and stlogs.day = lf.day,

                    and stlogs.hour = lf.hour

               where stlogs.filename = lf.filename

)

select count(res.timestamp) over (partition by res.year, res.month,

           res.day,

           res.hour,

           res.jsonurl) as cnt,

           filename,

           jsonurl,

           timestamp,

           year,

           month, day, hour

from res

""".format(...)

result = hive_ctx.sql(sql).repartition('year', 'month', 'day', hour')

Thus, Process log files -- DONE. We can run the process with azkaban on schedule (each hour)

Handle delayed data. The count should update as delayed log files are delivered.-- DONE. The delayed data will be processed

Reprocess log files without over counting requests. to re-process data we can just delete a partition of the required hour or day from dwh table, other requests will not be re-counted.

If there are any cases where system failure would result in incorrect counts. -- It can be fixed when the system will be recovered

Continue to handle growing scale of this data set over time. -- I would also advise moving outdated source data (oldest than 1 year for example into another dir, AWS provides cheaper storage type for such kind of data) .

This approach will work if we use table loader with property "append" but not "overwrite". If it were "overwrite" we would lose the previously loaded data into dwh table

Some additional notes:

Regarding the amount of data:

let's assume that we have not compressed format, so it means a single file with 10000 rows will have

apprx 0,5 MB size

the size of 1h delivery will be: 10000*0,5 = apprx 5 GB

the size of 24h delivery will be: 120 GB

the size of 1h delivery 10 x will be: 10000*0,5 * 10 = apprx 50 GB

the size of 24h delivery 10x will be: 1 TB 200 GB

the size of 1h delivery 100x will be: 10000*0,5 * 10 = apprx 500 GB

the size of 24h delivery 100x will be: 12 TB

We would advise running appropriate EMR cluster, and if data ingestion is 10x bigger? 100x bigger we can add new resources to it. So scaling can be resolved horizontally

There also will be pitfalls in the number of files. We will need a list of dropped files (new coming files). To generate it I would propose somehow appending the names of newcoming files with a separate process.

It will be cheaper to append rather than look through all of the files to get the list of the filenames. It can be Lambda function and further putting data to db.

I think that It is better to use just s3 for input file storage (but input file can be located in subdirs then spark will be able to deal with it scanning only the part of data).

Staging table can either created or not (it is a best practice of etl to create staging table). it is better to have staging (with AS-IS data) for quick debugging, data quality checks, easier data integration( we can use the same table in various custom processes).

In addition In the majority of cases customers prefer stability rather than rush.

As I mentioned I would track this data using the difference between files already loaded and the list of dropped files, but in this case, I will have to re-process a whole hour partition if a new file came.

Thus, when you need both scaling and speed, we would recommend you to re-process only some minimal delta using the minimum stack required.

For this particular case, we would draw your attention to EMR and would run spark batching app to handle the data. Azkaban can orchestrate this process