标签: mapreduce

有没有一种有效的方法来删除HBase中的多行或者我的用例气味是否不适合HBase？

有一个表格'chart',其中包含图表中的项目.行键的格式如下: chart|date_reversed|ranked_attribute_value_reversed|content_id

有时我想为给定日期重新生成图表,所以我想删除从'chart | date_reversed_1'开始直到'chart | date_reversed_2'的所有行.有没有比为Scan找到的每一行发出删除更好的方法？要删除的所有行都将彼此接近.

我需要删除行,因为我不希望一个项目(一个content_id)有多个条目,如果它的ranking_attribute_value已被更改,它将具有多个条目(其更改是图表需要重新生成的原因).

作为一个HBase的初学者,所以也许我可能会滥用行以获得更好的列 - 如果你有设计建议,很酷!或者,也许图表在文件中更好地生成(例如,没有HBase用于输出)？我正在使用MapReduce.

我有一个嵌套的JSON对象,用于下面的密钥状态:

{
"2011-01-19 09:41:00 AM": "Prototyping status application",
"2011-01-20 09:41:00 AM": "Playing with CouchDB"
}

这是一个小应用程序,用户可以输入他/她的状态.我希望从中获得最新状态.这种方法对于这样的应用程序是否有用,或者我必须有一个定义排序顺序的键？

获取最近日期的最佳方式是什么？

谢谢

我现在正在做一些数据分析测试,在第一个,非常简单,我有非常奇怪的结果.

这个想法如下:从互联网访问日志(每个访问的文档集合,测试90万个文档).我希望获得域的访问次数(MySQL中的GROUP BY),并获得10个访问最多的域

我在JavaScript中创建的脚本非常简单:

/* Counts each domain url */
m = function () {
    emit(this.domain, 1 );
}

r = function (key, values)    {
    total = 0;
    for (var i in values)    {
        total += Number(i);
    }

    return total;
}

/* Store of visits per domain statistics on NonFTP_Access_log_domain_visits collection */
res = db.NonFTP_Access_log.mapReduce(m, r, { out: { replace : "NonFTP_Access_log_domain_visits" } } );
db.NonFTP_Access_log_domain_visits.ensureIndex({ "value": 1});
db.NonFTP_Access_log_domain_visits.find({}).sort({ "value":-1 }).limit(10).forEach(printjson);

MySQL中的等价物是:

drop table if exists NonFTP_Access_log_domain_visits;
create table NonFTP_Access_log_domain_visits …

我坚持以下几点:

脚本返回任意数字n或数组,如下所示:

[["a"], ["b"], ["c"], ["d"]]

我需要使用promise循环遍历数组then(),但由于我不知道会有多少元素,所以我最终做到了这一点:

  var bundle_list = [["a"], ["b"], ["c"], ["d"]];

  var x = bundle_list.reduce(function(current, next) {
  console.log(current);

  // requestBundle will also return a promise
  return requestBundle(current)
    .then(function(bundle_response) {
      // do foo
      console.log("CALLING NEXT")
      console.log(next);
      return RSVP.resolve(next);
    });
})

x.then(function(last_response) {
  return console.log("DONE")
});

我的问题是我的reduce/map两个都在我的异步代码运行之前触发所有迭代,所以我得到3倍的current控制台,然后是done控制台.所以我的所有地图"循环"都会立即运行,结果会稍后(稍后)计时...

我正在使用这个RSVP实现,但它是A +所以不应该是一个问题.我一直在努力解决这里提供的答案,但我无法让它正常工作.

问题:
是否可以使用任意数量的then语句创建"then-chain" .如果是这样,一些指针表示赞赏!

谢谢!

我们已经有一个小火花集群运行了一个月,现在已成功执行作业或让我启动一个火花壳到集群.

如果我向集群提交作业或使用shell连接到它并不重要,错误总是相同的.

    root@~]$ $SPARK_HOME/bin/spark-shell
Spark assembly has been built with Hive, including Datanucleus jars on classpath
14/11/10 20:43:01 INFO spark.SecurityManager: Changing view acls to: root,
14/11/10 20:43:01 INFO spark.SecurityManager: Changing modify acls to: root,
14/11/10 20:43:01 INFO spark.SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(root, ); users with modify permissions: Set(root, )
14/11/10 20:43:01 INFO spark.HttpServer: Starting HTTP Server
14/11/10 20:43:01 INFO server.Server: jetty-8.y.z-SNAPSHOT
14/11/10 20:43:01 INFO server.AbstractConnector: Started SocketConnector@0.0.0.0:60223
14/11/10 20:43:01 INFO util.Utils: Successfully started …

目前我有两个HBase的表(让我们称之为tableA和tableB).使用单阶段MapReduce作业,tableA读取数据并将其保存到tableB.目前,两个表都位于同一HBase集群中.但是,我需要重新定位tableB到其群集上.

是否可以在Hadoop中配置单阶段映射reduce作业以从HBase的单独实例中读取和写入？

我试图在Oozie工作流程中聚合一些数据.但是聚合步骤失败.

我在日志中发现了两个兴趣点:第一个是错误(？)似乎反复出现:

容器完成后,它会被终止,但退出时返回非零退出代码143.

它结束了:

2015-05-04 15:35:12,013 INFO [IPC Server handler 7 on 49697] org.apache.hadoop.mapred.TaskAttemptListenerImpl: Progress of TaskAttempt attempt_1430730089455_0009_m_000048_0 is : 0.7231312
2015-05-04 15:35:12,015 INFO [IPC Server handler 19 on 49697] org.apache.hadoop.mapred.TaskAttemptListenerImpl: Progress of TaskAttempt attempt_1430730089455_0009_m_000048_0 is : 1.0

然后当它被Application Master杀死时:

2015-05-04 15:35:13,831 INFO [AsyncDispatcher event handler] org.apache.hadoop.mapreduce.v2.app.job.impl.TaskAttemptImpl: Diagnostics report from attempt_1430730089455_0009_m_000048_0: Container killed by the ApplicationMaster.
Container killed on request. Exit code is 143
Container exited with a non-zero exit code 143

第二个兴趣点是完全崩溃工作的实际错误,这发生在reduce阶段,不确定这两个是否相关:

2015-05-04 15:35:28,767 INFO [IPC Server handler 20 …

我试图乘以以镶木地板格式存储的大矩阵,所以小心不要将RDD存储在内存中,但是从镶木地板阅读器中得到OOM错误:

15/12/06 05:23:36 WARN TaskSetManager: Lost task 950.0 in stage 4.0   
(TID 28398, 172.31.34.233): java.lang.OutOfMemoryError: Java heap space
at org.apache.parquet.hadoop.ParquetFileReader$ConsecutiveChunkList.readAll(ParquetFileReader.java:755)
at org.apache.parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:494)
at org.apache.parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:127)
at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:208)
at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:201)
...

具体来说,矩阵是一个46752×54843120的32位浮点密集矩阵,以镶木地板格式存储(每行约为1.7GB未压缩).

下面的代码将此矩阵作为Spark IndexedRowMatrix加载并将其乘以随机向量(行与关联的字符串标签一起存储,并且浮点数必须转换为双精度,因为IndexedRows只能使用双精度数):

val rows = {
  sqlContext.read.parquet(datafname).rdd.map {
    case SQLRow(rowname: String, values: WrappedArray[Float]) =>
    // DenseVectors have to be doubles
      val vector = new DenseVector(values.toArray.map(v => v.toDouble)) 
      new IndexedRow(indexLUT(rowname), vector)
    }
}

val nrows : Long = 46752
val ncols = 54843120
val A = new IndexedRowMatrix(rows, nrows, ncols)
A.rows.unpersist() …

我正在使用spark来计算用户评论的页面,但是java.lang.StackOverflowError当我在大数据集上运行我的代码时,我会不断获得Spark (40k条目).当在少量条目上运行代码时,它工作正常.

输入示例:

product/productId: B00004CK40   review/userId: A39IIHQF18YGZA   review/profileName: C. A. M. Salas  review/helpfulness: 0/0 review/score: 4.0   review/time: 1175817600 review/summary: Reliable comedy review/text: Nice script, well acted comedy, and a young Nicolette Sheridan. Cusak is in top form.

代码:

public void calculatePageRank() {
    sc.clearCallSite();
    sc.clearJobGroup();

    JavaRDD < String > rddFileData = sc.textFile(inputFileName).cache();
    sc.setCheckpointDir("pagerankCheckpoint/");

    JavaRDD < String > rddMovieData = rddFileData.map(new Function < String, String > () {

        @Override
        public String call(String arg0) throws Exception {
            String[] data = arg0.split("\t"); …

我通读了权威指南和网上的其他一些链接,包括这里的链接

我的问题是

洗牌和分拣到底发生了什么？

根据我的理解,它们发生在映射器和缩减器上.但是有些链接提到了在映射器上进行重排并在减速器上进行排序.

有人可以确认我的理解是否正确; 如果没有,他们可以提供我可以通过的其他文件吗？