buildscript {
    repositories {
        jcenter()
        maven { url 'https://repo.spring.io/milestone' }
        maven { url 'https://plugins.gradle.org/m2/' }
    }

    dependencies {
        classpath "org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}"
        classpath "com.bmuschko:gradle-docker-plugin:5.3.0"
    }
}

subprojects {
    apply plugin: 'idea'
    apply plugin: 'java'
    apply plugin: 'org.springframework.boot'
    apply plugin: "io.spring.dependency-management"
    apply plugin: "com.bmuschko.docker-spring-boot-application"

    group = 'pers.henglin'
    version = '0.0.1-SNAPSHOT'
    sourceCompatibility = JavaVersion.VERSION_1_8

    jar.enabled = true
    bootJar.enabled = false

    repositories {
        jcenter()
        maven { url 'https://repo.spring.io/milestone' }
    }

    dependencies {
        compileOnly 'org.projectlombok:lombok'
        annotationProcessor 'org.projectlombok:lombok'
        testImplementation "org.springframework.boot:spring-boot-starter-test"
    }

    dependencyManagement {
        imports {
            mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
        }
    }
}

dependencies {
    implementation 'org.springframework.cloud:spring-cloud-starter-netflix-eureka-server'

    implementation 'org.springframework:springloaded:1.2.8.RELEASE'
    implementation 'org.springframework.boot:spring-boot-devtools'
}

docker {
    springBootApplication {
        baseImage = 'openjdk:jre-alpine'
        maintainer = 'heng-lin "enjoymycodinglife@gmail.com"'
        ports = [8761]
    }
}

public class BuyPhoneBean implements Writable {
    private String productName;
    private double productPrice;
    private String customerIds;

    public BuyPhoneBean(){
        super();
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(this.productName);
        out.writeDouble(this.productPrice);
        out.writeUTF(this.customerIds);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.productName = in.readUTF();
        this.productPrice = in.readDouble();
        this.customerIds = in.readUTF();
    }
    
    ...

}

public class BuyPhoneRecordReader extends RecordReader<Text, BuyPhoneBean> {
    private static Logger logger = LoggerFactory.getLogger(BuyPhoneRecordReader.class);

    private LineRecordReader lineRecordReader;

    private Text key = new Text();
    private BuyPhoneBean value = new BuyPhoneBean();

    private Text lineValue;

    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException {
        this.lineRecordReader = new LineRecordReader();
        this.lineRecordReader.initialize(split, context);
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if(lineRecordReader.nextKeyValue()){
            lineValue = lineRecordReader.getCurrentValue();
            byte[] line = lineValue.getBytes();
            int lineLen = lineValue.getLength();
            String data = new String(line, 0, lineLen);
            setKeyValue(this.key, this.value, data.split(REGEX_COLS_SPLIT_SYMBOL));
        }else{
            return false;
        }
        return true;
    }

    private void setKeyValue(Text key, BuyPhoneBean value, String[] cols){
        if(cols.length < 3){
            logger.warn("this cols[] length < 3");
            return;
        }
        String productName = cols[0];
        double productPrice = Double.parseDouble(cols[1]);
        String customerIds = cols[2];

        key.set(new Text(productName));

        value.setProductName(productName);
        value.setProductPrice(productPrice);
        value.setCustomerIds(customerIds);
    }

    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        return this.key;
    }

    @Override
    public BuyPhoneBean getCurrentValue() throws IOException, InterruptedException {
        return this.value;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return this.lineRecordReader.getProgress();
    }

    @Override
    public void close() throws IOException {
        this.lineRecordReader.close();
    }
}

/**
 * Expert users can override this method for more complete control over the
 * execution of the Mapper.
 * @param context
 * @throws IOException
 */
public void run(Context context) throws IOException, InterruptedException {
  setup(context);
  try {
    while (context.nextKeyValue()) {
      map(context.getCurrentKey(), context.getCurrentValue(), context);
    }
  } finally {
    cleanup(context);
  }
}

public class BuyPhoneInputFormat extends FileInputFormat<Text, BuyPhoneBean> {
    @Override
    public RecordReader<Text, BuyPhoneBean> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        BuyPhoneRecordReader recordReader = new BuyPhoneRecordReader();
        recordReader.initialize(split, context);
        return recordReader;
    }
}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    Configuration conf = new Configuration();
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

    if(otherArgs.length < 2){
        System.err.println("Usage: " + JOB_NAME + " <in> <out>");
        System.exit(2);
    }

    Job job = new HadoopJob(conf, JOB_NAME)
            .mapReduce(TestRecordReaderJob.class, SequenceMapper.class, SequenceReducer.class)
            .mapKeyValue(Text.class, BuyPhoneBean.class)
            .reducerKeyValue(Text.class, BuyPhoneBean.class)
            .getJob();

    job.setInputFormatClass(BuyPhoneInputFormat.class);
    job.setOutputFormatClass(SequenceFileOutputFormat.class);

    FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

    System.exit(job.waitForCompletion(true) ? 0 : 1);
}

public class CustomerBean implements Writable {
    protected static final String CUSTOMER = "customer.txt";
    protected static final String PRODUCT = "product.txt";
    protected static final String PRICE = "price.txt";
    
    private String dataType;
    
    private String customerId;
    private String productId;
    private String productName;
    private double productPrice;
    
    public CustomerBean() {
        super();
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(this.dataType);
        out.writeUTF(this.customerId);
        out.writeUTF(this.productName);
        out.writeDouble(this.productPrice);
    }
    
    @Override
    public void readFields(DataInput in) throws IOException {
        this.dataType = in.readUTF();
        this.customerId = in.readUTF();
        this.productName = in.readUTF();
        this.productPrice = in.readDouble();
    }
    
    ...

}

public class BuyPhoneMapper extends Mapper<LongWritable, Text, Text, CustomerBean> {
    private static Logger logger = LoggerFactory.getLogger(BuyPhoneMapper.class);
    
    private String dataType;
    private CustomerBean customerBean = new CustomerBean();
    private Text outputKey = new Text();
    
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        this.dataType = inputSplit.getPath().getName();
    }
    
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String record = value.toString();
        String[] cols = record.split(Constant.REGEX_COLS_SPLIT_SYMBOL);
        this.customerBean.setDataType(this.dataType);
        switch(this.dataType){
            case CUSTOMER:
                setCustomer(cols, this.outputKey, this.customerBean);
                break;
            case PRODUCT:
                setProduct(cols, this.outputKey, this.customerBean);
                break;
            case PRICE:
                setPrice(cols, this.outputKey, this.customerBean);
                break;
            default:
                logger.warn("It can't find this data type: {}", this.dataType);
                return;
        }
        context.write(this.outputKey, this.customerBean);
    }
    
    // record: customerId, productId
    private void setCustomer(String[] cols, Text key, CustomerBean customerBean){
        String customerId = cols[0];
        String productId = cols[1];
        customerBean.setCustomerId(customerId);
        customerBean.setProductId(productId);
        customerBean.setProductName(ENPTY_STRING);
        customerBean.setProductPrice(0d);
        key.set(productId);
    }
    
    // record: productId, productName
    private void setProduct(String[] cols, Text key, CustomerBean customerBean){
        String productId = cols[0];
        String productName = cols[1];
        customerBean.setCustomerId(ENPTY_STRING);
        customerBean.setProductId(productId);
        customerBean.setProductName(productName);
        customerBean.setProductPrice(0d);
        key.set(productId);
    }
    
    // record: productId, productPrice
    private void setPrice(String[] cols, Text key, CustomerBean customerBean){
        String productId = cols[0];
        double productPrice = Double.parseDouble(cols[1]);
        customerBean.setCustomerId(ENPTY_STRING);
        customerBean.setProductId(productId);
        customerBean.setProductName(ENPTY_STRING);
        customerBean.setProductPrice(productPrice);
        key.set(productId);
    }
}

public class BuyPhoneReducer extends Reducer<Text, CustomerBean, Text, Text> {
    private static Logger logger = LoggerFactory.getLogger(BuyPhoneReducer.class);
    
    private Text outputKey = new Text();
    private Text outputValue = new Text();
    
    @Override
    protected void reduce(Text key, Iterable<CustomerBean> values, Context context) throws IOException, InterruptedException {
        List<CustomerBean> customers = new ArrayList<>();
        String productName = ENPTY_STRING;
        double productPrice = 0d;
        for(CustomerBean customerBean:values){
            switch(customerBean.getDataType()){
                case CUSTOMER:
                    addCustomer(customers, customerBean);
                    break;
                case PRODUCT:
                    productName = customerBean.getProductName();
                    break;  
                case PRICE:
                    productPrice = customerBean.getProductPrice();
                    break;
                default:
                    logger.warn("It can't find this data type: {}", customerBean.getDataType());
                    return;
            }
        }
        writeResult(context, customers, productName, productPrice);
    }
    
    private void addCustomer(List<CustomerBean> customers, CustomerBean customer){
        CustomerBean temp = new CustomerBean();
        try {
            BeanUtils.copyProperties(temp, customer);
        } catch (IllegalAccessException | InvocationTargetException e) {
            e.printStackTrace();
        }
        customers.add(temp);
    }
    
    private void writeResult(Context context, List<CustomerBean> customers, String productName, double productPrice) throws IOException, InterruptedException {
        if(customers.isEmpty()){
            return;
        }
        List<String> customerIds = new ArrayList<>();
        for(CustomerBean customer:customers){
            customer.setProductName(productName);
            customer.setProductPrice(productPrice);
            customerIds.add(customer.getCustomerId());
        }
        this.outputKey.set(productName + REGEX_COLS_SPLIT_SYMBOL + productPrice);
        this.outputValue.set(Joiner.on(REGEX_DATA_JOIN_SPLIT_SYMBOL).join(customerIds));
        context.write(this.outputKey, this.outputValue);
    }
}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    Configuration conf = new Configuration();
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    
    if(otherArgs.length < 2){
        System.err.println("Usage: " + JOB_NAME + " <in> <out>");
        System.exit(2);
    }
    
    Job job = new HadoopJob(conf, JOB_NAME)
            .mapReduce(BuyPhoneJob.class, BuyPhoneMapper.class, BuyPhoneReducer.class)
            .mapKeyValue(Text.class, CustomerBean.class)
            .reducerKeyValue(Text.class, Text.class)
            .getJob();
    
    FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    
    System.exit(job.waitForCompletion(true) ? 0 : 1);
}