小编dra*_*nxo的帖子

我在完成问题时遇到了问题.hgetall(),这是我尝试过的:

Jedis jedis = new Jedis(REDIS_MASTER_NODE);
jedis.connect();
jedis.configSet("timeout", "30");

Map<String, String> alreadyStored = jedis.hgetAll(redisTargetHash);

这就是我得到的:

Exception in thread "main" redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketTimeoutException: Read timed out
    at redis.clients.jedis.Protocol.process(Protocol.java:79)
    at redis.clients.jedis.Protocol.read(Protocol.java:131)
    at redis.clients.jedis.Connection.getBinaryMultiBulkReply(Connection.java:199)
    at redis.clients.jedis.Jedis.hgetAll(Jedis.java:851)

这解决了这个问题:

Jedis jedis = new Jedis(REDIS_MASTER_NODE, 6379, 1800);

我想运行我在Mahout In Action中找到的代码:

package org.help;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.NamedVector;
import org.apache.mahout.math.VectorWritable;

public class SeqPrep {

    public static void main(String args[]) throws IOException{

        List<NamedVector> apples = new ArrayList<NamedVector>();

        NamedVector apple;

        apple = new NamedVector(new DenseVector(new double[]{0.11, 510, 1}), "small round green apple");        

        apples.add(apple);

        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path("appledata/apples");

        SequenceFile.Writer writer = new SequenceFile.Writer(fs,  conf, path, Text.class, …

我正在编写一个运行 aws emr 命令的 bash 脚本（aws emr 版本 1.5.2）。

我如何告诉我的脚本等到 emr 作业完成后再继续？该--wait-for-steps期权现在已折旧。

通过jq我得到了这个，但这似乎是错误的方法：

STEP_STATUS_STATE=$(aws emr list-steps --cluster-id ${CLUSTER_ID} | jq '.Steps[0].Status.State' | tr -d '"')
while [[ ${STEP_STATUS_STATE} == PENDING ]] || [[ ${STEP_STATUS_STATE} == RUNNING ]]; do
    STEP_STATUS_STATE=$(aws emr list-steps --cluster-id ${CLUSTER_ID} | jq '.Steps[0].Status.State' | tr -d '"')
    echo $(date) ${STEP_STATUS_STATE}
    sleep 10
done

我有几个依赖的.proto文件syntax = "proto3";.我还有一个Maven项目,用于构建Hadoop/Spark作业(Hadoop 2.7.1和Spark 1.5.2).我想在Hadoop/Spark中生成数据,然后根据我的proto3文件对其进行序列化.

使用libprotoc 3.0.0,我生成的Java源代码在我的Maven项目中工作正常,只要我在pom.xml中有以下内容:

<dependency>
  <groupId>com.google.protobuf</groupId>
  <artifactId>protobuf-java</artifactId>
  <version>3.0.0-beta-1</version>
</dependency>  

现在,当我在一个部署到集群的作业中使用我的libprotoc生成的类时,我遇到了:

java.lang.VerifyError : class blah overrides final method mergeUnknownFields.(Lcom/google/protobuf/UnknownFieldSet;)Lcom/google/protobuf/GeneratedMessage$Builder;
    at java.lang.ClassLoader.defineClass1(Native Method)
    at java.lang.ClassLoader.defineClass(ClassLoader.java:760)

ClassLoader失败似乎是合理的,因为Hadoop/Spark依赖于protobuf-java 2.5.0,这与我的3.0.0-beta-1不兼容.我还注意到protobufs(大概是版本<3)已经在其他几个地方进入我的罐子:

$ jar tf target/myjar-0.1-SNAPSHOT.jar | grep protobuf | grep '/$'
org/apache/hadoop/ipc/protobuf/
org/jboss/netty/handler/codec/protobuf/
META-INF/maven/com.google.protobuf/
META-INF/maven/com.google.protobuf/protobuf-java/
org/apache/mesos/protobuf/
io/netty/handler/codec/protobuf/
com/google/protobuf/
google/protobuf/

有什么我可以做的(Maven Shade？)来解决这个问题吗？

类似的问题:Spark java.lang.VerifyError

我有一个查询返回来自我们集群中分布的7个不同索引的〜200K命中.我将结果处理为:

while (true) {
    scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(600000)).execute().actionGet();

    for (SearchHit hit : scrollResp.getHits()){
            //process hit}

    //Break condition: No hits are returned
    if (scrollResp.hits().hits().length == 0) {
        break;
    }
}

我注意到client.prepareSearchScroll行可以在返回下一组搜索命中之前挂起很长时间.我运行代码的时间越长,这似乎越来越糟糕.

我的搜索设置是:

SearchRequestBuilder searchBuilder = client.prepareSearch( index_names )
    .setSearchType(SearchType.SCAN)
    .setScroll(new TimeValue(60000)) //TimeValue?
    .setQuery( qb )
    .setFrom(0) //?
    .setSize(5000); //number of jsons to get in each search, what should it be? I have no idea.
    SearchResponse scrollResp = searchBuilder.execute().actionGet();

在检查许多结果时,预计扫描和滚动需要很长时间吗？我对弹性搜索很新,所以请记住,我可能会遗漏一些非常明显的东西.

我的查询:

QueryBuilder qb = QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("tweet", interesting_words));

在解析了一些jsons后,我有一个数组的单列DataFrame

scala> val jj =sqlContext.jsonFile("/home/aahu/jj2.json")
res68: org.apache.spark.sql.DataFrame = [r: array<bigint>]
scala> jj.first()
res69: org.apache.spark.sql.Row = [List(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)]

我想把每一行分成几行.怎么样？

编辑:

原始的json文件:

{"r": [0,1,2,3,4,5,6,7,8,9]}
{"r": [0,1,2,3,4,5,6,7,8,9]}

我想要一个20行的RDD或DataFrame.

我不能简单地在这里使用flatMap - 我不确定spark中的相应命令是什么:

scala> jj.flatMap(r => r)
<console>:22: error: type mismatch;
 found   : org.apache.spark.sql.Row
 required: TraversableOnce[?]
              jj.flatMap(r => r)

我有一个带有城市名称和GPS坐标的文本文件(UTF-8,~50K行).示例行:

San Pedro    locality   -3367   -5968   Argentina       Buenos Aires    San Pedro
Talagante    locality   -3366   -7093   Chile   Metropolitana   Talagante
Peñaflor     locality   -3362   -7092   Chile   Metropolitana   Talagante

第三和第四列是最后一列中城市的GPS坐标.

给定GPS坐标,我需要找到壁橱城市.我需要做数亿次这样的事情.有哪些工具可以帮助我完成这项任务？Java/Python解决方案将是理想的选择.

我想用一个自定义Comparator的TreeMultiSet。这是我所做的：

public static class OccurenceComparator implements Comparator<Multiset.Entry<String>>{
    @Override
    public int compare(Multiset.Entry<String> e1, Multiset.Entry<String> e2) {
        return e2.getCount() - e1.getCount() ;
    }
}

public static void main(String args[]) throws Exception{
    Comparator<Multiset.Entry<String>> occurenceComparator = new OccurenceComparator();
    Multiset<String> treeMultiSet = TreeMultiset.create(occurenceComparator);
}

这是我得到的：

未解决的编译问题：create(Comparator<? super E>)类型中的方法TreeMultiset不适用于参数 (Comparator<Multiset.Entry<String>>)

我难住了

我对Hadoop的理解是,每个计算节点上的并行性是通过为每个核启动单独的jvms来实现的.

我观察到每个jvm拥有数十个线程,每个节点导致数千个线程.我想不出任何产生这么多线程的理由.这是怎么回事？

例如,这是一个简单的猪脚本,用于解析和过滤一些jsons:

/*
 * Get tweets with GPS
 */
REGISTER $JAR;

json_eb = LOAD '$IN_DIRS' USING com.twitter.elephantbird.pig.load.JsonLoader('-nestedLoad') as (json:map[]);

--parse json with twitter's library
parsed0 = FOREACH json_eb GENERATE  STRSPLIT(json#'id',':').$2 AS tweetId:chararray,
                                    STRSPLIT(json#'actor'#'id',':').$2 AS userId:chararray,
                                    json#'postedTime' AS postedTime:chararray,
                                    json#'geo'#'coordinates' AS gps:chararray;
parsed1 = FILTER parsed0 BY (gps IS NOT NULL);

STORE parsed1 INTO '$OUT_DIR' USING PigStorage();

我运行这个脚本,mapred用户在我的节点上启动了33个进程(我有32个核心):

rfcompton@node19 ~> ps -u mapred | grep -v PID | wc -l
33

看顶部:

  PID USER      PR  NI  VIRT  RES  SHR S …