我是Cascading / Hadoop的新手,正在尝试以本地模式(即在内存中)运行一个简单示例。该示例仅复制一个文件: import java.util.Properties; import cascading.flow.Flow; import cascading.flow.FlowConnector; import cascading.flow.FlowDef; import cascading.flow.local.LocalFlowConnector; import cascading.pipe.Pipe…

2020年11月29日 0条评论 17点热度 阅读全文

我最近不得不做级联工作。我以前从未有过使用分布式系统的经验,因此在理解如何进行这项工作时遇到了一些麻烦。 我有一个配置文件,其中有很多存储桶: Bucket{ bucket_name: "X" input_path: "s3://..." key_column: 1 value_column: 2 multivalue: false default_value: type_column: int } ... 基本上,我要做的就是使用它来收集一堆文件(每个文件都像一个tsv表,它将url键映射到某个值),然后按url…

2020年11月21日 0条评论 27点热度 阅读全文

我的数据具有以下结构: +data |-2014080700_00.txt |-2014080700_01.txt |-2014080701_00.txt |- ... |-2014080723_00.txt |-2014080800_00.txt |- ... |-2014090800_00.txt 我知道我可以将数据目录中的所有文件与 Tap一起使用,如下所示: Tap inTap = new Hfs( new TextLine(), "/path/to/data"); 但是我想要目录的特定部分,例如仅日期 2…

2020年10月27日 0条评论 10点热度 阅读全文

在map-reduce中,我将提取输入文件名,如下所示 public void map(WritableComparable<Text> key, Text value, OutputCollector<Text,Text> output, Reporter reporter) throws IOException { FileSplit fileSplit = (FileSplit)reporter.getInputSplit(); String filename = fileSplit.…

2020年10月21日 0条评论 7点热度 阅读全文

我需要在hadoop上进行级联处理UTF-16LE编码的文件。我尝试了以下方法,但这些方法均无效。 将值-Xmx1024m -Dfile.encoding=UTF-16LE分配给中的属性mapreduce.map.java.opts mapred-site.xml 由于NullPointerException而在以下位置失败:com.google.common.base.Preconditions.checkNotNull(Preconditions.java:187),但是此方法适用于UTF-8。 hadoop无…

2020年10月9日 0条评论 18点热度 阅读全文

我正在学习级联。现在,我正在其官方网站上查找有关工作计数示例的第二个教程。我从中复制代码并尝试运行,但始终会出现以下错误: Exception in thread "main" cascading.flow.planner.PlannerException: could not build flow from assembly: [[token][com.starscriber.cascadingtest.Main.main(Main.java:44)] unable to resolve argument sel…

2020年9月26日 0条评论 16点热度 阅读全文

我在Cloudera-5(CDH-5)VM中使用驱动的自我托管版本。 我能够成功安装驱动服务器,并能够在URL localhost.localdomain:8080中打开服务器。 我在文件cascading-service.properties中提供了以下值 cascading.management.document.service.apikey=API_KEY from Driven Server cascading.management.service.jar=/usr/lib/cascading-2.5.5/…

2020年7月22日 0条评论 15点热度 阅读全文

String inputPath = args[0]; FileSystem dfs = new DistributedFileSystem(); FileStatus[] files= null; try{ files = dfs.listStatus(new path(inputPath)); } catch(IOExcpeption err){ //Do stuff } 使用maven可以很好地构建代码。但是,当我尝试运行它时,我在try子句中得到了nullPointerException。有任何想法吗? 解…

2020年7月17日 0条评论 18点热度 阅读全文

有人可以解释为什么当我尝试在本地文件系统上运行作业时此代码引发异常吗?我是Hadoop / Cascading的新手,正在尝试每行获取多个列并计算这些列中值的频率(不区分它们来自哪个列键) 我很茫然。如果有人可以向我解释我的思维过程误入歧途,我将不胜感激。我更关心学习技术及其库,而不是编写工作代码。 这是源代码: http://pastebin.com/USTUJ5Da 这是堆栈跟踪: SLF4J:无法加载类“org.slf4j.impl.StaticLoggerBinder”。SLF4J:默认为无操作(NOP)记…

2020年7月9日 0条评论 18点热度 阅读全文

我想知道一种情况,我可以实现cascading.operation.Buffer,cascading.operation.Function并在元组上执行操作。 public class Hello extends BaseOperation implements Buffer, Function { } 我不确定以上类(class)是否有效。 如果有效,那么将调用哪种操作方法? 我应该在哪里实现任何实际方案?有性能问题吗? 解决方案如下: 您正在谈论的接口(interface)是: cascading.operat…

2020年6月22日 0条评论 17点热度 阅读全文