解决方法:
1、使用Linux命令,如下:
String tmpFile = "/tmp" + File.separator + TableUtil.TABLE_PREFIX_MAILING_MEMBER + "@" + mailListId + ".log"; String scp_command = String.format("scp %s x.x.x.x:%s", fileKey, tmpFile); logger.info("Current TaskId#" + taskId + " : " + scp_command); Process process = Runtime.getRuntime().exec(scp_command); logger.info("Current TaskId#" + taskId + ", SCP_COMMAND : <OUTPUT>"); InputStreamReader inputStream = new InputStreamReader(process.getInputStream()); BufferedReader inputReader = new BufferedReader(inputStream); String line = null; while((line = inputReader.readLine()) != null) { logger.info(line); } logger.info("Current TaskId#" + taskId + ", SCP_COMMAND : </OUTPUT>"); if (inputReader != null) { inputReader.close(); } if (inputStream != null) { inputStream.close(); } logger.info("Current TaskId#" + taskId + ", SCP_COMMAND : <ERROR>"); InputStreamReader errorStream = new InputStreamReader(process.getErrorStream()); BufferedReader errorReader = new BufferedReader(errorStream); line = null; while((line = errorReader.readLine()) != null) { logger.info(line); } logger.info("Current TaskId#" + taskId + ", SCP_COMMAND : </ERROR>"); if (errorReader != null) { errorReader.close(); } if (errorStream != null) { errorStream.close(); } process.waitFor(); logger.info("Current TaskId#" + taskId + " : " + scp_command); Thread.sleep(5000); // 赋权限 String chown_command = String.format("%s '%s %s'", "ssh x.x.x.x", "chown hadoop:hadoop -R", tmpFile); logger.info("Current TaskId#" + taskId + " :" + chown_command); process = Runtime.getRuntime().exec(chown_command); logger.info("Current TaskId#" + taskId + ", CHOWN_COMMAND : <OUTPUT>"); inputStream = new InputStreamReader(process.getInputStream()); inputReader = new BufferedReader(inputStream); line = null; while((line = inputReader.readLine()) != null) { logger.info(line); } logger.info("Current TaskId#" + taskId + ", CHOWN_COMMAND : </OUTPUT>"); if (inputReader != null) { inputReader.close(); } if (inputStream != null) { inputStream.close(); } logger.info("Current TaskId#" + taskId + ", CHOWN_COMMAND : <ERROR>"); errorStream = new InputStreamReader(process.getErrorStream()); errorReader = new BufferedReader(errorStream); line = null; while((line = errorReader.readLine()) != null) { logger.info(line); } logger.info("Current TaskId#" + taskId + ", CHOWN_COMMAND : </ERROR>"); if (errorReader != null) { errorReader.close(); } if (errorStream != null) { errorStream.close(); } process.waitFor(); logger.info("Current TaskId#" + taskId + " : " + chown_command);
2、使用如下方法:
public static void copyLocalFileHdfs(String localPath, String hdfsPath) throws IOException { Configuration conf = new Configuration(); conf.set("fs.defaultFS", "hdfs://Ucluster"); conf.set("dfs.nameservices", "Ucluster"); conf.set("dfs.ha.namenodes.Ucluster", "xx,yy"); conf.set("dfs.namenode.rpc-address.Ucluster.xx","hostname:port"); conf.set("dfs.namenode.rpc-address.Ucluster.yy","hostname:port"); conf.set("dfs.client.failover.proxy.provider.Ucluster", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"); try { FileSystem fs = FileSystem.get(conf); Path src = new Path(localPath); Path dst = new Path(hdfsPath); fs.copyFromLocalFile(src, dst); } catch (Exception e) { e.printStackTrace(); } }
相关推荐
项目实战:Java一站式解决Hive内用Spark取数,新建ES索引,灌入数据,并且采用ES别名机制,实现ES数据更新的无缝更新,底层采用Spark计算框架,数据较快。
4.使用Sqoop将数据从Hive导入MySQL 5.使用Sqoop将数据从MySQL导入HBase 6.使用HBase Java API把数据从本地导入到HBase中 7.使用R对MySQL中的数据进行可视化分析 内含三份报告和数据集,报告中有源码,是用spark做的
使用Spark框架进行网站用户购物分析 ...4、使用Sqoop将数据从Hive导入MySQL 5、使用Sqoop将数据从MySQL导入HBase 6、使用HBase Java API把数据从本地导入到HBase中 7、使用R对MySQL中的数据进行可视化分析
OSM2Hive是Hive的用户定义功能的集合,以允许OSM XML数据导入。 它以更易于使用的格式读取Hive表中的XML文件,并对其进行解析以创建新表。 应用程序测试使用JUnit 4框架。 用法 必须在Hive中直接调用OSM2Hive。 为此...
通过一定的java程序设计做数据解析、结构化、各自合并成一个文件即可方便load到hive中。 输出:(两张表,一个用户表,一个评论内容表) 将两个类别下各自对应的文件集,进行解析、结构化。 通过load语法,将数据...
1.6.4 导入数据 1.6.5 表的修改 1.6.6 表的丢弃 1.7 查询数据 1.7.1 排序(Sorting)和聚集(Aggregating) 1.7.2 MapReduce脚本 1.7.3 连接 1.7.4 子查询 1.7.5 视图(view) 1.8 用户定义函数(User-Defined ...
1.6.4 导入数据 1.6.5 表的修改 1.6.6 表的丢弃 1.7 查询数据 1.7.1 排序(Sorting)和聚集(Aggregating) 1.7.2 MapReduce脚本 1.7.3 连接 1.7.4 子查询 1.7.5 视图(view) 1.8 用户定义函数(User-...
总项目0,大数据生态解决方案数据平台数据平台:bdp-dataplatform:基于大数据、数据平台、微服务、机器学习、商城、自动化运维、DevOps、容器部署平台、数据平台采集、数据平台存储、数据平台计算、数据平台开发、...
如何优化应用的性能,包括前端优化、Java程序优化、数据库查询优化等。 如何进行Java应用故障的在线排查,包括一系列排查工具的使用,以及一些实际案例的介绍等。 4.1 在线日志分析 220 4.1.1 日志分析常用...
《Hadoop实战》作为云计算所青睐的分布式架构,Hadoop是一个用Java语言实现的软件框架,在由大量计算机组成的集群中运行海量数据的分布式计算,是谷歌实现云计算的重要基石。《Hadoop实战》分为3个部分,深入浅出地...
通过统一的UI,类似于工作流的图形拖放开发体验可满足数据应用程序开发的整个生命周期,包括数据导入,脱敏清理,数据分析,数据挖掘,质量检查,可视化,计划到数据输出应用程序等。 。 借助Linkis的连接,可重用...
7.1.2 MongoDB的数据导入与导出 121 7.2 面向列数据库中数据schema的演进 124 7.3 HBase数据导入与导出 125 7.4 键/值存储中的数据演变 126 7.5 小结 126 第8章 数据索引与排序 127 8.1 数据库索引的基本概念...
本资源为大数据基础到中高级教学资源,适合稍微有点大数据或者java基础的人群学习,资源过大,上传乃是下载链接,不多说,上目录: 1_java基础2 l3 a2 a$ t7 J2 b+ `- p 2_java引入ide-eclipse 3_java基础知识-循环...
2011年10月版次:1.1开本:16开装帧:平装字数:417千字页数:253页内容简介编辑作为云计算所青睐的分布式架构,Hadoop是一个用Java语言实现的软件框架,在由大量计算机组成的集群中运行海量数据的分布式计算,是...