小编tri*_*oid的帖子

Hadoop 2.6不支持开箱即用的s3a,所以我尝试了一系列解决方案和修复,包括:

部署与hadoop-aws和aws-java-sdk =>无法读取环境变量的凭据将hadoop-aws添加到maven =>各种传递依赖冲突

有没有人成功地使两者都有效？

我正在EC2集群上部署Spark数据处理作业,集群的作业很小(16个内核总共有120G RAM),最大的RDD只有76k +行.但是在中间严重偏斜(因此需要重新分区),并且每行在序列化后具有大约100k的数据.这项工作总是陷入重新分配的困境.也就是说,该作业将不断出现以下错误并重试:

org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle

org.apache.spark.shuffle.FetchFailedException: Error in opening FileSegmentManagedBuffer

org.apache.spark.shuffle.FetchFailedException: java.io.FileNotFoundException: /tmp/spark-...

我试图找出问题,但似乎机器的内存和磁盘消耗都会导致这些错误低于50%.我也尝试了不同的配置,包括:

let driver/executor memory use 60% of total memory.
let netty to priortize JVM shuffling buffer.
increase shuffling streaming buffer to 128m.
use KryoSerializer and max out all buffers
increase shuffling memoryFraction to 0.4

但它们都不起作用.小作业总是触发相同系列的错误并最大限度地重试(最多1000次).在这种情况下如何排除故障？

如果你有任何线索,非常感谢.

我试图测试以下代码但没有成功:

class TestClass
{
  private class ND2Customer
  {
    public String name;
    public String description;
    public String email;
    public Boolean multiuser;

    public String dnszone;
    public String uri;
    public String type;

    public ND2Customer()
    {

    }
  }

  @Test
  public void TestJackson() throws JsonParseException, JsonMappingException, IOException
  {
    String json="{\"description\": \"test1u\", \"dnszone\": \"test1.public.sevenltest.example.com.\", \"uri\": \"http://199.127.129.69/customer/test1\", \"multiuser\": true, \"type\": \"2.0.3-3146\", \"email\": \"test1@com.com\", \"name\": \"test1\"}";
    ObjectMapper mapper = new ObjectMapper();

    ND2Customer casted=mapper.readValue(json, ND2Customer.class);

    String castedback=mapper.defaultPrettyPrintingWriter().writeValueAsString(casted);
    System.out.println(castedback);
  } 
}

这个问题与此不同: 用Jackson反序列化JSON - 为什么JsonMappingException"没有合适的构造函数"？

这一个: JsonMappingException:找不到类型[simple type,class]的合适构造函数:无法从JSON对象实例化

这一个: JsonMappingException:找不到类型[simple …

众所周知,Spark中的分区器对任何"广泛"操作都会产生巨大的性能影响,因此通常会在操作中进行自定义.我正在尝试以下代码:

val rdd1 =
  sc.parallelize(1 to 50).keyBy(_ % 10)
    .partitionBy(new HashPartitioner(10))
val rdd2 =
  sc.parallelize(200 to 230).keyBy(_ % 13)

val cogrouped = rdd1.cogroup(rdd2)
println("cogrouped: " + cogrouped.partitioner)

val unioned = rdd1.union(rdd2)
println("union: " + unioned.partitioner)

我看到默认情况下cogroup()总是会生成带有自定义分区程序的RDD,但union()不会,它将始终恢复为默认值.这是违反直觉的,因为我们通常假设PairRDD应该使用其第一个元素作为分区键.有没有办法"强制"Spark合并2个PairRDD以使用相同的分区键？

我想在我的java程序中使用这个android依赖项:

http://jcenter.bintray.com/com/o3dr/android/dronekit-android/2.9.0/

所以我将所有插件和存储库添加到我的maven pom文件中:

<repositories>
    <repository>
        <id>central</id>
        <name>bintray</name>
        <url>http://jcenter.bintray.com</url>
        <!--<snapshots>-->
            <!--<enabled>false</enabled>-->
        <!--</snapshots>-->
    </repository>
</repositories>

<pluginRepositories>
    <pluginRepository>
        <!--<snapshots>-->
            <!--<enabled>false</enabled>-->
        <!--</snapshots>-->
        <id>central</id>
        <name>bintray-plugins</name>
        <url>http://jcenter.bintray.com</url>
    </pluginRepository>
</pluginRepositories>

<build>
    <plugins>
        <plugin>
            <groupId>com.simpligility.maven.plugins</groupId>
            <artifactId>android-maven-plugin</artifactId>
            <version>4.1.0</version>

            <extensions>true</extensions>
            <configuration>
                <sign>
                    <debug>false</debug>
                </sign>
            </configuration>
        </plugin>
        ...

然后我将依赖项添加到同一个文件中:

    <dependency>
        <groupId>com.o3dr.android</groupId>
        <artifactId>dronekit-android</artifactId>
        <version>2.9.0</version>
        <type>aar</type>
    </dependency>

但没有任何反应,maven忽略它就好像它不存在,如果我在scala文件中导入它的主包并用它构建它mvn clear install -U,我得到这个错误:

[ERROR] /home/peng/git-drone/dronespike/src/main/scala/Main.scala:1:object o3dr不是包com
[ERROR] import com.o3dr.android._
[ERROR] ^的成员

问题:我该怎么做才能解决这个问题？

Scala案例类具有默认的toString函数.但是当这个case类使用现有的toString()函数扩展一个特征时,它将变得无用.我该如何防止这种情况？

我在EMR上的火花程序经常出现这个错误:

Caused by: javax.net.ssl.SSLPeerUnverifiedException: peer not authenticated
    at sun.security.ssl.SSLSessionImpl.getPeerCertificates(SSLSessionImpl.java:421)
    at org.apache.http.conn.ssl.AbstractVerifier.verify(AbstractVerifier.java:128)
    at org.apache.http.conn.ssl.SSLSocketFactory.connectSocket(SSLSocketFactory.java:397)
    at org.apache.http.impl.conn.DefaultClientConnectionOperator.openConnection(DefaultClientConnectionOperator.java:148)
    at org.apache.http.impl.conn.AbstractPoolEntry.open(AbstractPoolEntry.java:149)
    at org.apache.http.impl.conn.AbstractPooledConnAdapter.open(AbstractPooledConnAdapter.java:121)
    at org.apache.http.impl.client.DefaultRequestDirector.tryConnect(DefaultRequestDirector.java:573)
    at org.apache.http.impl.client.DefaultRequestDirector.execute(DefaultRequestDirector.java:425)
    at org.apache.http.impl.client.AbstractHttpClient.execute(AbstractHttpClient.java:820)
    at org.apache.http.impl.client.AbstractHttpClient.execute(AbstractHttpClient.java:754)
    at org.jets3t.service.impl.rest.httpclient.RestStorageService.performRequest(RestStorageService.java:334)
    at org.jets3t.service.impl.rest.httpclient.RestStorageService.performRequest(RestStorageService.java:281)
    at org.jets3t.service.impl.rest.httpclient.RestStorageService.performRestHead(RestStorageService.java:942)
    at org.jets3t.service.impl.rest.httpclient.RestStorageService.getObjectImpl(RestStorageService.java:2148)
    at org.jets3t.service.impl.rest.httpclient.RestStorageService.getObjectDetailsImpl(RestStorageService.java:2075)
    at org.jets3t.service.StorageService.getObjectDetails(StorageService.java:1093)
    at org.jets3t.service.StorageService.getObjectDetails(StorageService.java:548)
    at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrieveMetadata(Jets3tNativeFileSystemStore.java:172)
    at sun.reflect.GeneratedMethodAccessor18.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:190)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:103)
    at org.apache.hadoop.fs.s3native.$Proxy8.retrieveMetadata(Unknown Source)
    at org.apache.hadoop.fs.s3native.NativeS3FileSystem.getFileStatus(NativeS3FileSystem.java:414)
    at org.apache.hadoop.fs.FileSystem.exists(FileSystem.java:1398)
    at org.apache.hadoop.fs.s3native.NativeS3FileSystem.create(NativeS3FileSystem.java:341)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:906)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:887)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:784)

我做了一些研究,发现通过设置环境变量,可以在低安全性情况下禁用此身份验证:

com.amazonaws.sdk.disableCertChecking=true

但是我只能用spark-submit.sh --conf设置它,它只影响驱动程序,而大多数错误都在工作者身上.

有没有办法将它们传播给工人？

非常感谢.

假设我有一个Generic超类:

class GenericExample[T](
                         a: String,
                         b: T
                       ) {

  def fn(i: T): T = b
}

和一个具体的子类:

case class Example(
                    a: String,
                    b: Int
                  ) extends GenericExample[Int](a, b)

我想通过scala反射得到函数"fn"的类型参数,所以我选择并过滤其成员:

import ScalaReflection.universe._

val baseType = typeTag[Example]

val member = baseType
  .tpe
  .member(methodName: TermName)
  .asTerm
  .alternatives
  .map(_.asMethod)
  .head

    val paramss = member.paramss
    val actualTypess: List[List[Type]] = paramss.map {
      params =>
        params.map {
          param =>
            param.typeSignature
        }
    }

我期待scala给我正确的结果List(List(Int)),相反,我只得到了通用List(List(T))

通过文档进行处理我发现typeSignature是罪魁祸首:

 *  This method always returns signatures in the most generic …

至于Spark 1.6+,支持自定义分区位置的唯一API是在创建RDD时:

  /** Distribute a local Scala collection to form an RDD, with one or more
    * location preferences (hostnames of Spark nodes) for each object.
    * Create a new partition for each collection item. */
  def makeRDD[T: ClassTag](seq: Seq[(T, Seq[String])]): RDD[T]

尽管在某些情况下非常有用(例如,当RDD.compute()必须访问一些本地资源时,而不仅仅是HDFS).这是暴露这种设置的唯一地方,但在第一次洗牌之后它将很快被丢弃(下游分区将从其最大的父级继承preferredLocation)

  // snippet from org.apache.spark.rdd.ShuffledRDD.scala
  override protected def getPreferredLocations(partition: Partition): Seq[String] = {
    val tracker = SparkEnv.get.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
    val dep = dependencies.head.asInstanceOf[ShuffleDependency[K, V, C]]
    tracker.getPreferredLocationsForShuffle(dep, partition.index)
  }

和cogroup/join(其中下游分区将使用具有显式分区程序的第一个父级).

我想知道这个设计是否是故意的,或者存在更好的托管解决方案.您认为为shuffled/coalesced/cogrouped/join RDD指定preferredLocation的最佳解决方案是什么？我是否必须编写自己的RDD实例才能实现此目的？

非常感谢您的洞察力.

更新我推测了一个可能的解决方案,令我惊讶的是不起作用:

在Apache Spark cogroup中,如何确保不移动1个> 2个操作数的RDD？

所以我删除了答案,如果你有一些有用的东西,欢迎你在这里分享,否则我们将不得不等待https://issues.apache.org/jira/browse/SPARK-18078

当 scalatest 中的相等断言失败时。它通常只显示不同的部分，例如：

"...error;
!I e: F[Arg]
[g invalid because
nonconformant bounds;
[Arg, Nothing]
[A <: __wrapper$1$47213a912399466a973eddce7b3420f4.__wrapper$1$47213a912399466a973eddce7b3420f4.]Bounds.Base, B]
  im..." did not equal "...error;
!I e: F[Arg]
[Bounds.g invalid because
nonconformant bounds;
[Arg, Nothing]
[A <: ]Bounds.Base, B]
  im..."
org.scalatest.exceptions.TestFailedException: "...error;
!I e: F[Arg]
[g invalid because
nonconformant bounds;
[Arg, Nothing]
[A <: __wrapper$1$47213a912399466a973eddce7b3420f4.__wrapper$1$47213a912399466a973eddce7b3420f4.]Bounds.Base, B]
  im..." did not equal "...error;
!I e: F[Arg]
[Bounds.g invalid because
nonconformant bounds;
[Arg, Nothing]
[A <: ]Bounds.Base, B]
  im..."

这份报告中丢失了很多信息。另外，当IDE配备diff解析器时，它会错误地显示比较结果。有没有办法在scalatest中禁用此功能，这样报告中就不会出现任何省略号？