【数据科学】Pig Design Patterns

The HBase ingress and egress pattern

1. The egress code
   
2. The following code illustrates content storage of a Pig relation into HBase table:
   
3. 
   
4. /*
   
5. Load the transactions dataset using PigStorage into the relation transactions
   
6. */
   
7. transactions = LOAD '/user/cloudera/pdp/datasets/hbase/transactions.csv' USING PigStorage( ',' ) AS (
   
8.     listing_id: chararray,
   
9.     transaction_date: chararray,
   
10.     customer_id: int,
    
11.     age: chararray,
    
12.     residence_area: chararray,
    
13.     product_subclass: int,
    
14.     product_id: long,
    
15.     amount: int,
    
16.     asset: int,
    
17.     sales_price: int);
    
18. 
    
19. /*
    
20. * Some processing logic goes here which is deliberately left out to improve readability
    
21. */
    
22. 
    
23. /*
    
24. Use HBaseStorage to store data from the Pig relation transactions into a HBase table hbase://retail_transactions.
    
25. The individual contents of transactions are mapped to three column families transaction_details, product_details and customer_details.
    
26. */
    
27. STORE transactions INTO 'hbase://retail_transactions' USING org.apache.pig.backend.hadoop.hbase.HBaseStorage('transaction_details:transaction_date customer_details:customer_id customer_details:age customer_details:residence_area product_details:product_subclass product_details:product_id product_details:amount product_details:asset product_details:sales_price');

复制代码

The Hive ingress and egress patterns

1. Importing data using RCFile
   
2. 
   
3. The following code illustrates the usage of HiveColumnarLoader that loads data from a Hive table stored in a RCFile:
   
4. 
   
5. /*
   
6. Register the Piggybank jar file to be able to use the UDFs in it
   
7. */
   
8. REGISTER '/usr/share/pig/contrib/piggybank/java/piggybank.jar';
   
9. 
   
10. -- Register Hive common and exec jars
    
11. REGISTER '/usr/lib/hive/lib/hive-common-0.11.0.1.3.0.0-107.jar';
    
12. REGISTER '/usr/lib/hive/lib/hive-exec-0.11.0.1.3.0.0-107.jar';
    
13. 
    
14. /*
    
15. Load retail_transactions_rc  RCfile and specify the names of the columns of the table and their types in the constructor of HiveColumnarLoader.
    
16. */
    
17. transactions = LOAD '/apps/hive/warehouse/transactions_db.db/retail_transactions_rc' USING org.apache.pig.piggybank.storage.HiveColumnarLoader('transaction_no int,transaction_date string,cust_no int,amount double,category string,product string,city string,state string,spendby string');
    
18. 
    
19. /*
    
20. * Some processing logic goes here which is deliberately left out to improve readability
    
21. */
    
22. 
    
23. /*
    
24. Display the contents of the relation transactions on the console
    
25. */
    
26. DUMP transactions;

复制代码

The Hive ingress and egress patterns

1. Importing data using HCatalog
   
2. 
   
3. The following code illustrates the loading of data from Hive using HCatalog:
   
4. 
   
5. /*
   
6. Specify the table name as the input to the HCatLoader function provided by HCatalog.
   
7. This function abstracts the storage location, files type, schema from the user and takes only the table name as input
   
8. */
   
9. transactions = LOAD 'transactions_db.retail_transactions' USING org.apache.hcatalog.pig.HCatLoader();
   
10. 
    
11. /*
    
12. * Some processing logic goes here which is deliberately left out to improve readability
    
13. */
    
14. 
    
15. /*
    
16. Display the contents of the relation transactions on the console
    
17. */
    
18. DUMP transactions;

复制代码

The Hive ingress and egress patterns

1. The following code illustrates the egression of data to Hive using HCatStorer:
   
2. 
   
3. -- Register piggybank and hcatalog-pig-adapter jars
   
4. REGISTER '/usr/share/pig/contrib/piggybank/java/piggybank.jar';
   
5. REGISTER '/usr/lib/hcatalog/share/hcatalog/hcatalog-pig-adapter.jar';
   
6. 
   
7. /*
   
8. Load the transactions dataset into the relation transactions
   
9. */
   
10. transactions = LOAD '/user/cloudera/pdp/datasets/hive/retail_transactions.csv' USING org.apache.pig.piggybank.storage.CSVLoader() AS (transaction_no:int, transaction_date:chararray, cust_no:int, amount:double, category:chararray, product:chararray, city:chararray, state:chararray, spendby:chararray);
    
11. 
    
12. /*
    
13. * Some processing logic goes here which is deliberately left out to improve readability
    
14. */
    
15. 
    
16. /*
    
17. Specify the Hive table name transactions_db.retail_transactions as the input to the HCatStorer function.
    
18. The contents of the relation transactions are stored into the Hive table.
    
19. */
    
20. STORE transactions INTO 'transactions_db.retail_transactions' using org.apache.hcatalog.pig.HCatStorer();

复制代码

The mainframe ingestion pattern

1. The following is a Java code snippet of VSAMLoader, which is a custom loader implementation:
   
2. 
   
3. @Override
   
4. public ResourceSchema getSchema(String arg0, Job arg1) throws IOException {
   
5.   .
   
6.   .
   
7.   while (it.hasNext()) {
   
8.     Map.Entry pairs = (Map.Entry) it.next();
   
9.     //Get the next key/value pairs
   
10.     String key = (String) pairs.getKey();
    
11.     String value = (String) pairs.getValue();
    
12.     /*For Group and Alphanumeric types in copybook, return
    
13.     pig compliant type chararray*/
    
14.     if (value.toString()
    
15.     .equals("class net.sf.cb2java.copybook.Group")
    
16.     || value.toString().equals("class net.sf.cb2java.copybook.AlphaNumeric")){
    
17.        fieldSchemaList.add(new FieldSchema(key,
    
18.         org.apache.pig.data.DataType.CHARARRAY));
    
19.       }
    
20.     /*For Decimal type in copybook, return
    
21.     pig compliant type integer*/
    
22.     else if (value.toString()
    
23.     .equals("class net.sf.cb2java.copybook.Decimal")){
    
24.       fieldSchemaList.add(new FieldSchema(key,
    
25.         org.apache.pig.data.DataType.INTEGER));
    
26.     }
    
27.     // Else return default bytearray
    
28.     else
    
29.     {
    
30.        fieldSchemaList.add(new FieldSchema(key,
    
31.        org.apache.pig.data.DataType.BYTEARRAY));
    
32.     }
    
33.     }
    
34.   return new ResourceSchema(new Schema(fieldSchemaList));
    
35. }

复制代码

1. The code for simple JSON
   
2. 
   
3. The code to load JSON files using JsonLoader is shown as follows:
   
4. 
   
5. /*
   
6. Use JSONLoader UDF, it takes in the parameter of the JSON schema and loads the contents of the JSON file emails.json into a map enron_emails
   
7. */
   
8. enron_emails = LOAD '/user/cloudera/pdp/datasets/json/emails.json' USING JsonLoader('body:chararray, from:chararray, tos:chararray, ccs:chararray, bccs:chararray, date:chararray, message_id:chararray, subject:chararray');
   
9. 
   
10. /*
    
11. * Some processing logic goes here which is deliberately left out to improve readability
    
12. */
    
13. 
    
14. /*
    
15. Display the contents of the relation enron_emails on the console
    
16. */
    
17. DUMP enron_emails;
    
18. It is important to note that the JsonLoader does not use the AS clause to supply the schema.

复制代码

1. The code for nested JSON
   
2. 
   
3. The Pig script to load nested JSON is shown as follows, and we use the elephant-bird libraries to accomplish this:
   
4. 
   
5. /*
   
6. Register elephant-bird and JSON jar files
   
7. */
   
8. REGISTER '/home/cloudera/pdp/jars/elephant-bird-core-3.0.5.jar';
   
9. REGISTER '/home/cloudera/pdp/jars/elephant-bird-pig-3.0.5.jar';
   
10. REGISTER '/home/cloudera/pdp/jars/json-simple-1.1.1.jar';
    
11. 
    
12. /*
    
13. Use ElephantBird's JsonLoader for loading a nested JSON file
    
14. The parameter –nestedload denotes nested loading operation
    
15. */
    
16. emails = LOAD '/user/cloudera/pdp/datasets/json/emails.json' USING com.twitter.elephantbird.pig.load.JsonLoader('-nestedLoad');
    
17. 
    
18. /*
    
19. * Some processing logic goes here which is deliberately left out to improve readability
    
20. */
    
21. 
    
22. /*
    
23. Display the contents of the relation emails on the console
    
24. */
    
25. DUMP emails;

复制代码

1. The egress code
   
2. The following section shows the code and its explanation to egress data stored in a Pig relation to JSON format:
   
3. 
   
4. /*
   
5. Load the JSON file using JsonLoader to the relation enron_emails
   
6. */
   
7. enron_emails = LOAD '/user/cloudera/pdp/datasets/json/emails.json' USING JsonLoader('body:chararray, from:chararray, tos:chararray, ccs:chararray, bccs:chararray, date:chararray, message_id:chararray, subject:chararray');
   
8. 
   
9. /*
   
10. * Some processing logic goes here which is deliberately left out to improve readability
    
11. */
    
12. 
    
13. /*
    
14. Use JsonStorage to store the contents of the relation to a json file
    
15. */
    
16. STORE enron_emails into '/user/cloudera/pdp/output/json/output.json' USING JsonStorage()

复制代码

1. Pig script
   
2. The following is the Pig script illustrating the implementation of this pattern:
   
3. 
   
4. /*
   
5. Register the datatypeinferer and custom storage jar files
   
6. */
   
7. REGISTER '/home/cloudera/pdp/jars/datatypeinfererudf.jar';
   
8. REGISTER'/home/cloudera/pdp/jars/customdatatypeinfererstorage.jar';
   
9. 
   
10. /*
    
11. Load the transactions dataset into the relation transactions
    
12. */
    
13. transactions = LOAD'/user/cloudera/pdp/datasets/data_profiling/transactions.csv'USING  PigStorage(',') AS (transaction_id:long,transaction_date:chararray, cust_id:chararray, age:chararray,area:chararray, prod_subclass:int, prod_id:long, amt:int,asset:int, sales_price:int, phone_no:chararray,country_code:chararray);
    
14. 
    
15. /*
    
16. Infer the data type of the field cust_id by invoking the DataTypeInfererUDF.
    
17. It returns a tuple with the inferred data type.
    
18. */
    
19. data_types = FOREACH transactions GENERATEcom.profiler.DataTypeInfererUDF(cust_id) AS inferred_data_type;
    
20. 
    
21. /*
    
22. Compute the count of each data type, total count, percentage.
    
23. The data type with the highest count is considered as dominant data type
    
24. */
    
25. grpd = GROUP data_types BY inferred_data_type;
    
26. inferred_type_count = FOREACH grpd GENERATE group ASinferred_type, COUNT(data_types) AS count;
    
27. grpd_inf_type_count_all = GROUP inferred_type_count ALL;
    
28. total_count = FOREACH grpd_inf_type_count_all GENERATESUM(inferred_type_count.count) AS tot_sum,MAX(inferred_type_count.count) AS max_val;
    
29. percentage = FOREACH inferred_type_count GENERATE inferred_type AStype, count AS total_cnt,CONCAT((Chararray)ROUND(count*100.0/total_count.tot_sum),'%') ASpercent,(count==total_count.max_val?'Dominant':'Other') ASinferred_dominant_other_datatype;
    
30. percentage_ord = ORDER percentage BYinferred_dominant_other_datatype ASC;
    
31. 
    
32. /*
    
33. CustomDatatypeInfererStorage UDF extends the StoreFunc. All the abstract methods have been overridden to implement logic that writes the contents of the relation into a file in a custom report like format.
    
34. The results are stored on the HDFS in the directory datatype_inferer
    
35. */
    
36. STORE percentage_ord INTO'/user/cloudera/pdp/output/data_profiling/datatype_inferer'using com.profiler.CustomDatatypeInfererStorage('cust_id','chararray');

复制代码