Issue in Hadoop FTPFileSystem

[Dhanashri Desai <dh...@gmail.com>]

I am trying to write data to FTP server using normal syntax of credetials:
ftp://user:password@host/outputFilePath.. Below is my cascading code

Code:
package com.ftp.readwrite;
import java.io.IOException;
import cascading.flow.FlowDef;
import cascading.flow.hadoop.HadoopFlowConnector;
import cascading.operation.aggregator.Count;
import cascading.operation.regex.RegexSplitGenerator;
import cascading.pipe.Each;
import cascading.pipe.Every;
import cascading.pipe.GroupBy;
import cascading.pipe.Pipe;
import cascading.scheme.hadoop.TextLine;
import cascading.tap.SinkMode;
import cascading.tap.Tap;
import cascading.tap.hadoop.Hfs;
import cascading.tuple.Fields;


public class MainClass {

    @SuppressWarnings("deprecation")
    public static void main(String args[]) throws IOException
    {

        Tap source=new Hfs(new TextLine(new Fields("line")),
"C:\\Users\\DhanashriD\\Desktop\\test.txt");
        Tap sink=new Hfs(new TextLine(new Fields("line1")), "
ftp://test:test@10.30.125.63/output",SinkMode.REPLACE);


        Pipe line=new Pipe("Lines");
        line=new Each(line,new RegexSplitGenerator("\\s+"));
        Pipe group=new GroupBy(line);

         Pipe counts=new Every(group, new Count());

         FlowDef flowdef=FlowDef.flowDef().addSource(line,
source).addTailSink(counts, sink);

         new HadoopFlowConnector().connect(flowdef).complete();

    }

}

I am getting following error:

java.io.IOException: Cannot rename parent(source):
ftp://test:test@10.30.125.63/op/_temporary/_attempt_local_0001_r_000000_0,
parent(destination):  ftp://test:test@10.30.125.63/output
    at org.apache.hadoop.fs.ftp.FTPFileSystem.rename(FTPFileSystem.java:547)
    at org.apache.hadoop.fs.ftp.FTPFileSystem.rename(FTPFileSystem.java:512)
    at
org.apache.hadoop.mapred.FileOutputCommitter.moveTaskOutputs(FileOutputCommitter.java:154)
    at
org.apache.hadoop.mapred.FileOutputCommitter.moveTaskOutputs(FileOutputCommitter.java:172)
    at
org.apache.hadoop.mapred.FileOutputCommitter.commitTask(FileOutputCommitter.java:132)
    at
org.apache.hadoop.mapred.OutputCommitter.commitTask(OutputCommitter.java:221)
    at org.apache.hadoop.mapred.Task.commit(Task.java:1001)
    at org.apache.hadoop.mapred.Task.done(Task.java:871)

Please help to resolve this error


Thanks
Dhanashri