/*Owner & Copyrights: Vance King Saxbe. A.*//* Copyright (c) <2014> Author Vance King Saxbe. A, and contributors Power Dominion Enterprise, Precieux Consulting and other contributors. Modelled, Architected and designed by Vance King Saxbe. A. with the geeks from GoldSax Consulting and GoldSax Technologies email @vsaxbe@yahoo.com. Development teams from Power Dominion Enterprise, Precieux Consulting. Project sponsored by GoldSax Foundation, GoldSax Group and executed by GoldSax Manager.*//*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the
 * NOTICE file distributed with this work for additional information regarding copyright ownership. The ASF
 * licenses this file to you under the Apache License, Version 2.0 (the "License"); you may not use this file
 * except in compliance with the License. You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and limitations under the License.
 */

package com.goldsaxfoundation.bigdata.Module11;

import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.pig.*;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.*;
import org.apache.pig.impl.logicalLayer.schema.Schema;

import java.io.IOException;
import java.util.*;

/**
 * TypedCommonLogLoader is used to load logs based on Apache's
 * Common Log Format (CLF), based on the following format:
 * <p/>
 * "%h %l %u %t \"%r\" %>s %b"
 * <p/>
 * Example of log line:
 * <p/>
 * 127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] "GET /apache_pb.gif HTTP/1.0" 200 2326
 * <p/>
 * Example use of LoadFunc:
 * <p/>
 * raw = LOAD 'common_log.txt' USING com.manning.hip.ch7.TypedCommonLogLoader;
 * <p/>
 * describe raw;
 * logs: {remoteAddr: chararray,remoteLogname: chararray,userid: chararray,
 * time: chararray,requestLine: chararray,statusCode: long,objSize: long,
 * method: chararray,resource: chararray,protocol: chararray,epoch: long}
 */

public class ComplexTupleLoader extends FileInputLoadFunc
    implements LoadMetadata {

 protected RecordReader reader = null;
  private TupleFactory tupleFactory = TupleFactory.getInstance();
  private BagFactory bagFactory = BagFactory.getInstance();

  @Override
  public Tuple getNext() throws IOException {
    try {
      if(!reader.nextKeyValue()) {
        return null;
      }
    } catch (InterruptedException e) {
      int errCode = 6018;
      String errMsg = "Error while reading input";
      throw new ExecException(errMsg, errCode,
        PigException.REMOTE_ENVIRONMENT, e);
    }    ArrayList<Object> tuple = new ArrayList<Object>();
    tuple.add("127.0.0.1");

    Map<String, Object> header = new HashMap<String, Object>();
    header.put("User-Agent", "Mozilla");
    tuple.add(header);

    ArrayList<Tuple> bodyTuples = new ArrayList<Tuple>();
    bodyTuples.add(newBodyTuple("keyword1"));
    bodyTuples.add(newBodyTuple("keyword2"));
    tuple.add(bagFactory.newDefaultBag(bodyTuples));

    return tupleFactory.newTuple(tuple);
  }

  public Tuple newBodyTuple(String line) {
    return tupleFactory.newTuple(Arrays.asList(line));
  }

  @Override
  public void setLocation(String location, Job job)
      throws IOException {
    FileInputFormat.setInputPaths(job, location);
  }

  @SuppressWarnings("rawtypes")
  @Override
  public InputFormat getInputFormat() throws IOException {
    return new TextInputFormat();
  }

  @Override
  public void prepareToRead(
      @SuppressWarnings("rawtypes") RecordReader reader,
      PigSplit split)
      throws IOException {
    this.reader = reader;
  }

  public ResourceSchema getSchema(String location, Job job)
      throws IOException {

    Schema schema = new Schema();

    Schema headerSchema = new Schema(
        Arrays.asList(
            new Schema.FieldSchema("word", DataType.CHARARRAY)
        ));

    schema.add(new Schema.FieldSchema("ip", DataType.CHARARRAY));
    schema.add(new Schema.FieldSchema("header", DataType.MAP));
    schema.add(new Schema.FieldSchema("keywords", headerSchema));
    return new ResourceSchema(schema);
  }

  public ResourceStatistics getStatistics(String location, Job job)
      throws IOException {
    return null;
  }

  public String[] getPartitionKeys(String location, Job job)
      throws IOException {
    return null;
  }

  public void setPartitionFilter(Expression partitionFilter)
      throws IOException {
  }
}
/*Packages, Programs, Algos for Mapreduce, Machine learning, ETL, as applicable in ERP & Financial domains. Modelled, Architected and designed by Vance King Saxbe. A. with the geeks from GoldSax Consulting, GoldSax Money, GoldSax Treasury, GoldSax Finance, GoldSax Banking and GoldSax Technologies email @vsaxbe@yahoo.com. Development teams from Power Dominion Enterprise, Precieux Consulting. Project sponsored by GoldSax Foundation, GoldSax Group and executed by GoldSax Manager.*//*email to provide support at vancekingsaxbe@powerdominionenterprise.com, businessaffairs@powerdominionenterprise.com, For donations please write to fundraising@powerdominionenterprise.com*/