Pig in the Cloud (запускаємо Pig скрипт на AWS)

В попередньому пості я описувала, як порахувати подібність між користувачами  на Pig. Основна перевага використання Pig - MapReduce завдання створяться і виконаються на основі вашого Pig скрипта. Використання технології MapReduce виправдане тільки у випадку розподілених обчислень, тобто потрібно мати або Hadoop кластер з реальних машин, або орендувати хмарний кластер. Сьогодні я опишу як підняти кластер на Amazon Web Services та запустити розподілений обрахунок подібності користувачів.



Для цього потрібно:
Ввійти в обліковий запис на AWS ( http://aws.amazon.com/) або створити новий

1. Перейти на сторінку AWS security credentials
2. Закрити спливаюче вікно з інформацією про IAM
3. Розгорнути пункт Access Keys
4. Натиснути Create Access Key  Your access key (access key ID and secret access key) has been created successfully. 
5. Натиснути на Download Key File і запам'ятати, куди файл завантажився.

Згенерувати ключі доступу до EC2 через SSH

1. Найпростіший варіант(більше інформації тут ): 
2. Перейти на сторінку https://console.aws.amazon.com/
3. Обрати  EC2
4. В розділі Resources знайти пункт Key Pairs та клікнути по ньому
5. Натиснути Create Key Pair
6. Вказати ім'я для ключа
7. Перемістіть ключ .pem де ви плануєте його використовувати
8. Змініть права доступу 

 chmod 600 pig_blog.pem  

Завантажуємо в S3 файл small_transactions.csv який містить перших 3000 записів з transaction.csv:

1. Переходимо на https://console.aws.amazon.com/s3/home
2. Обираємо Create Bucket 
3. Вказуємо назву(напр. kaggleshoppers) 
4. Create 
5. Переходимо в новий бакет 
6. Upload 
7. Add files 
8. казуємо шлях до small_transactions.csv 
9. Start upload

Запускаємо кластер:

1. Переходимо на https://console.aws.amazon.com/elasticmapreduce/home
2. Натискаємо Create Cluster
3. Вказуємо ім'я кластера, напр. SimilarityCounter
4. Відключаємо логування (Logging)
5. В розділі Software Configuration список AMI version визначає, яка версію Hadoop буде на вашому кластері. Обираємо AMI 2.4.2 з Hadoop 1.0.3.
6. В розділі Security and Access в полі EC2 key pair вказуємо ключ, який було створено вище
7. Натискаємо на кнопку Create Cluster

На наступній сторінці відобразиться конфігурація вашого кластера та стан, у якому він знаходиться. Спочатку це "Strating". Потрібно дочекатись, поки воно зміниться на "Running".
Тепер ви можете приєднатись до вашого кластера, викорстовуючи команду:

 $ ssh -o "ServerAliveInterval 10" -i </path/to/saved/keypair/file.pem> hadoop@<master.public-dns-name.amazonaws.com>   

де 
- шлях до *.pem файла, вказаного вище  значення поля Master Public DNS на сторінці з описом кластеру в пункті хх.

 trans1 = LOAD 's3n://kaggleshoppers/small_transactions.csv' using PigStorage(',') AS (id:chararray, chain:chararray, dept:chararray, category:chararray, company:chararray, brand:chararray, date:chararray, productsize:int, productmeasure:chararray, purchasequantity:chararray, purchaseamount:chararray);  
 trans2 = LOAD 's3n://kaggleshoppers/small_transactions.csv' using PigStorage(',') AS (id:chararray, chain:chararray, dept:chararray, category:chararray, company:chararray, brand:chararray, date:chararray, productsize:int, productmeasure:chararray, purchasequantity:chararray, purchaseamount:chararray);  

Далі основна частина аналогічна описаній в попередньому пості

 trans1 = FILTER trans1 BY id != '"id"';  
 brand_1 = FOREACH trans1 GENERATE id, brand;  
 brand_1 = DISTINCT brand_1;  
 trans2 = FILTER trans2 BY id != '"id"';  
 brand_2 = FOREACH trans2 GENERATE id, brand;  
 brand_2 = DISTINCT brand_2;  
 grouped_brand = GROUP brand_1 BY id;  
 aug_brand = FOREACH grouped_brand GENERATE FLATTEN(brand_1) AS (id, brand), COUNT(brand_1) AS id_out;  
 grouped_brand_2 = GROUP brand_2 BY id;  
 aug_brand_2 = FOREACH grouped_brand_2 GENERATE FLATTEN(brand_2) AS (id, brand), COUNT(brand_2) AS id_out;  
 brand_joined = JOIN aug_brand BY brand, aug_brand_2 BY brand;  
 intersection = FOREACH brand_joined {  
          --  
          -- results in:  
          -- (X, Y, |X| + |Y|)  
          --   
          added_size = aug_brand::id_out + aug_brand_2::id_out;  
          GENERATE  
           aug_brand::id AS user1,  
           aug_brand_2::id AS user2,  
           added_size AS added_size  
          ;  
         };  
 intersect_grp  = GROUP intersection BY (user1, user2);  
 intersect_sizes = FOREACH intersect_grp {  
           --  
           -- results in:  
           -- (X, Y, |X /\ Y|, |X| + |Y|)  
           --  
           intersection_size = (double)COUNT(intersection);  
           GENERATE  
            FLATTEN(group)        AS (user1, user2),  
            intersection_size      AS intersection_size,  
            MAX(intersection.added_size) AS added_size -- hack, we only need this one time  
           ;  
          };  
 similarities = FOREACH intersect_sizes {  
          --  
          -- results in:  
          -- (X, Y, |X /\ Y|/|X U Y|)  
          --  
          similarity = (double)intersection_size/((double)added_size-(double)intersection_size);  
          GENERATE  
           user1     AS user1,  
           user2     AS user2,  
           similarity AS similarity  
          ;  
         };  
 DUMP similarities;  
 similarities = FILTER similarities BY user1 != user2 and similarity > 0;   

І зберігаємо результат в S3:

 STORE similarities into 's3n://kaggleshoppers/small_similarity_results' USING PigStorage(',');  

Якщо всі операції виконано успішно, можна завантажувати оригінальний файл transactions.csv розміром 20 Гб. Міняємо в скріпті вище 1 та 3 частини і коментуємо пункт DUMP similarities;
Конфігурація по замовчуванню:
Master:Running1m1.large

Core:Running2m1.small

не є оптимальною для нашого вихідного завдання (порахувати подібність попарно між 300 000 користувачів). Використання більшої кількості більших та оптимізованих під розрахунки інстансів та використання також спот інстансів дозволяє скоротити час обрахунку (а відповідно й потрачені суми).
Після закінчення роботи скрипта  кластер потрібно обов'язково зупинити.
Попередження.Повторення цього скрипта на AWS буде коштувати грошей (небагато, близько $1 для обробки 3000 записів про покупки, оскільки всі вищеперераховані операції займуть до години часу).
Tip Якщо ви навчаєтесь (не обов'язково у ВУЗі, це можуть бути і Coursera чи Udacity) то можете подати заявку на навчальний грант у розмірі $100 (взнати більше про гранти можна тут)

Приклад використання Pig для аналізу логів від Amazon(зараз інтерфейс трохи відрізняється, але розібратись можна)