小编oge*_*gen的帖子

我们的堆栈由Google Data Proc(Spark 2.0)和Google BigTable(HBase 1.2.0)组成,我正在寻找使用这些版本的连接器.

对于我找到的连接器,我不清楚Spark 2.0和新的DataSet API支持:

• spark-hbase:https://github.com/apache/hbase/tree/master/hbase-spark
• spark-hbase-connector:https://github.com/nerdammer/spark-hbase-connector
• hortonworks-spark/shc:https://github.com/hortonworks-spark/shc

该项目使用SBT在Scala 2.11中编写.

谢谢你的帮助

我想在我的本地开发机器上运行一个简单的Spark作业（通过Intellij），从Amazon s3读取数据。

我的build.sbt文件：

scalaVersion := "2.11.12"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "2.3.1",
  "org.apache.spark" %% "spark-sql" % "2.3.1",
  "com.amazonaws" % "aws-java-sdk" % "1.11.407",
  "org.apache.hadoop" % "hadoop-aws" % "3.1.1"
)

我的代码段：

val spark = SparkSession
    .builder
    .appName("test")
    .master("local[2]")
    .getOrCreate()

  spark
    .sparkContext
    .hadoopConfiguration
    .set("fs.s3n.impl","org.apache.hadoop.fs.s3native.NativeS3FileSystem")

  val schema_p = ...

  val df = spark
    .read
    .schema(schema_p)
    .parquet("s3a:///...")

我得到以下异常：

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/hadoop/fs/StreamCapabilities
    at java.lang.ClassLoader.defineClass1(Native Method)
    at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
    at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
    at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
    at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
    at java.security.AccessController.doPrivileged(Native …

我是Scala的新手,我想用相同的性能级别翻译我的Java代码.

给定n个浮点向量和一个附加向量,我必须计算所有n个点的乘积并获得最大值.

使用Java对我来说非常简单

public static void main(String[] args) {

    int N = 5000000;
    int R = 200;
    float[][] t = new float[N][R];
    float[] u = new float[R];

    Random r = new Random();

    for (int i = 0;i<N;i++) {
        for (int j = 0;j<R;j++) {
            if (i == 0) {
                u[j] = r.nextFloat();
            }
            t[i][j] = r.nextFloat();
        }
    }

    long ts = System.currentTimeMillis();
    float maxScore = -1.0f;

    for (int i = 0;i < N;i++) {
        float score = 0.0f;
        for (int …

我对akka-http很陌生，在同一条路线上并行运行查询时遇到麻烦。

我有一条路线可能会非常快（如果已缓存）或没有（大量CPU多线程计算）返回结果。我想并行运行这些查询，以防一小段查询经过一长段计算之后又到达，我不希望第二个调用等待第一个调用完成。

但是，如果这些查询在同一路由上，则似乎不会并行运行（如果在不同的路由上，则并行运行）

我可以在一个基本项目中复制它：

并行调用服务器3次（使用http：// localhost：8080 / test上的3个Chrome选项卡），响应分别达到3.0秒，6.0秒和9.0秒。我想查询不会并行运行。

在具有jdk 8的Windows 10上的6核（带有HT）计算机上运行。

build.sbt

name := "akka-http-test"

version := "1.0"

scalaVersion := "2.11.8"

libraryDependencies += "com.typesafe.akka" %% "akka-http-experimental" % "2.4.11"

* AkkaHttpTest.scala **

import java.util.concurrent.Executors

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import akka.stream.ActorMaterializer

import scala.concurrent.{ExecutionContext, Future}

object AkkaHttpTest extends App {

  implicit val actorSystem = ActorSystem("system") // no application.conf here
  implicit val executionContext = 
                             ExecutionContext.fromExecutor(Executors.newFixedThreadPool(6))
  implicit val actorMaterializer = ActorMaterializer()

  val route = path("test") {
    onComplete(slowFunc()) { slowFuncResult =>
      complete(slowFuncResult) …

我有一个由akka-http. 在某些情况下，我需要从外部数据库 ( Apache HBase)获取数据，如果数据库传送数据的时间过长，我希望查询失败。

一种天真的方法是将调用包装在 a 中Future，然后使用Await.result具有所需持续时间的an 阻塞它。

import scala.concurrent.duration._
import scala.concurrent.{Await, Future}

object AsyncTest1 extends App {

  val future = Future {
    getMyDataFromDB()
  }

  val myData = Await.result(future, 100.millis)
}

这似乎效率低下，因为此实现需要两个线程。有没有一种有效的方法来做到这一点？

我还有另一个用例，我想并行发送多个查询，然后以相同的延迟限制聚合结果。

val future1 = Future {
    getMyDataFromDB1()
  }

  val future2 = Future {
    getMyDataFromDB2()
  }

  val foldedFuture = Future.fold(
    Seq(future1, future2))(MyAggregatedData)(myAggregateFunction)
  )

  val myData = Await.result(foldedFuture, 100.millis)

同样的问题，实现这一点的最有效方法是什么？

谢谢你的帮助

我想知道一个不可变的scala集合(例如immutable.HashMap)是否是线程安全的.内容可能显然不会改变,但优化的内部可变状态对我来说还不清楚.

假设键和值是不可变的

谢谢你的帮助

只是阅读Scala集合实现,并注意到一个微妙的差异:

• 不可变的HashMap:class HashMap[A, +B]()
• 可变的HashMap:class HashMap[A, B]()

你能解释一下为什么不可变的值类型是协变的而可变的值不是吗？

谢谢你的帮助

我是akka-http的新手,我想用任意数量的行来传输一个csv.

例如,我想回复:

a,1
b,2
c,3

使用以下代码

implicit val actorSystem = ActorSystem("system")
implicit val actorMaterializer = ActorMaterializer()

val map = new mutable.HashMap[String, Int]()
map.put("a", 1)
map.put("b", 2)
map.put("c", 3)
val `text/csv` = ContentType(MediaTypes.`text/csv`, `UTF-8`)
val route =
  path("test") {
    complete {
      HttpEntity(`text/csv`, ??? using map)
    }
  }
Http().bindAndHandle(route,"localhost",8080)

谢谢你的帮助

编辑:感谢Ramon J Romero和Vigil

package test


import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpCharsets.`UTF-8`
import akka.http.scaladsl.model._
import akka.http.scaladsl.server.Directives._
import akka.stream._
import akka.util.ByteString

import scala.collection.mutable

object Test{

  def main(args: Array[String]) {

    implicit val actorSystem = ActorSystem("system") …