add hdfs source and sink implementation