小编Yas*_*r S的帖子

我试图在Eclipse中使用spark运行简单的以下代码:

import org.apache.spark.sql.SQLContext
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object jsonreader {  
  def main(args: Array[String]): Unit = {
    println("Hello, world!")
    val conf = new SparkConf()
      .setAppName("TestJsonReader")
      .setMaster("local")
      .set("spark.driver.memory", "3g") 
    val sc = new SparkContext(conf)

    val sqlContext = new SQLContext(sc)
    val df = sqlContext.read.format("json").load("text.json")

    df.printSchema()
    df.show   
  }
}

但是,我收到以下错误:

16/08/18 18:05:28 ERROR SparkContext: Error initializing SparkContext.
java.lang.IllegalArgumentException: System memory 259522560 must be at least 471859200. Please increase heap size using the --driver-memory option or spark.driver.memory in Spark configuration.

我按照这样的不同教程:如何设置Apache Spark Executor内存.大部分时间我使用 …

在 API 管理中，我创建了一个 API，可以调用无服务器函数应用程序。现在我想自动部署这个功能。以下是我在互联网上看到的可能性：

• 通过门户创建和配置api管理（这不是我所说的自动部署）
• 使用 Powershell 命令（不幸的是我正在使用 linux）
• ARM（Azure Resource Manager）：这并不容易，我确实找到了如何使用 Azure 函数应用创建 API
• Terraform：与 ARM 相同，我不清楚如何使用 Azure 函数应用创建 API

如果有人有经验、链接或想法，我将不胜感激。

问候，

我有一个带有来自我的物联网应用程序的JSON数据的Kafka经纪人。我从Spark Streaming应用程序连接到此服务器，以便进行一些处理。

我想将json数据的某些特定字段保存在内存（RAM）中，我相信我可以使用cache()和persist()运算符来实现。

下次当我在Spark Streaming应用程序中收到新的JSON数据时，我检入内存（RAM）中是否有可检索的公共字段。如果是，我做了一些简单的计算，最后更新了保存在内存（RAM）中的字段的值。

因此，我想知道以前描述的内容是否可行。如果是，我是否必须使用cache（）或persist（）？以及如何从内存中检索字段？

我正在开发一个关于 Kubernetes 的项目，在该项目中我使用 Spark SQL 创建表，并且我想将分区和模式添加到 Hive Metastore。但是，我没有找到任何在 Kubernetes 上安装 Hive Metastore 的正确文档。知道我已经安装了 PostGreSQL 数据库是否有可能？如果是的话，您能帮我提供任何官方文件吗？

提前致谢。

我使用自签名证书通过 TLS安装了 Minio（我使用helm在 Kubernetes 中安装了 Minio）。之前我可以在没有 TLS 的情况下使用 Minio 运行我的 Spark 作业。现在无法连接到 Minio（正常！）

然后，我从 tls 证书创建了一个信任库文件

keytool -import \
  -alias tls \
  -file tls.crt \
  -keystore truststore.jks \
  -storepass "$minioTruststorePass" \
  -noprompt

我使用信任库的内容创建一个 Kubernetes 密钥，并在 Spark-defaults.conf 中使用以下选项来让 Spark 使用信任库：

spark.kubernetes.driver.secrets.minio-truststore-secret

最后，我在我的spark-defaults.conf中做了以下所有更改，但同样的问题

spark.hadoop.fs.s3a.endpoint                                      https://smart-agriculture-minio:9000
spark.hadoop.fs.s3.awsAccessKeyId                                 <s3aAccessKey>
spark.hadoop.fs.s3.awsSecretAccessKey                             <s3aSecretKey>
spark.hadoop.fs.s3.impl                                           org.apache.hadoop.fs.s3a.S3AFileSystem
spark.hadoop.fs.s3a.access.key                                    <s3aAccessKey>
spark.hadoop.fs.s3a.secret.key                                    <s3aSecretKey>
spark.hadoop.fs.s3a.path.style.access                             true
spark.hadoop.fs.s3a.impl                                          org.apache.hadoop.fs.s3a.S3AFileSystem
spark.hadoop.fs.s3a.connection.ssl.enabled                        true
spark.driver.extraJavaOptions                                      -Djavax.net.ssl.trustStore=/opt/spark/conf/minio/truststore/truststore.jks -Djavax.net.ssl.trustStorePassword=<minioTruststorePass>
spark.executor.extraJavaOptions                                   -Djavax.net.ssl.trustStore=/opt/spark/conf/minio/truststore/truststore.jks -Djavax.net.ssl.trustStorePassword=<minioTruststorePass>

您是否曾经遇到过这个问题并且您有解决办法吗？

谢谢

我想使用 Spark Streaming 从 Kafka 检索数据。现在，我想将数据保存在远程 HDFS 中。我知道我必须使用函数 saveAsText。但是，我不知道如何准确指定路径。

如果我写这个是正确的：

myDStream.foreachRDD(frm->{
    frm.saveAsTextFile("hdfs://ip_addr:9000//home/hadoop/datanode/myNewFolder");
});

ip_addr我的 hdfs 远程服务器的 IP 地址在哪里。 /home/hadoop/datanode/是我安装hadoop时创建的DataNode HDFS目录（不知道要不要指定这个目录）。而且， myNewFolder是我要保存数据的文件夹。

提前致谢。

亚西尔

我正在使用Spark Streaming v2.0.0从Kafka检索日志并进行一些操作。我正在使用该功能mapWithState以保存和更新与设备相关的某些字段。我想知道此功能如何在群集中工作。确实，到目前为止，我只是使用独立模式，但稍后将在Yarn群集中尝试使用它。

但是，假设我有一个包含多个节点的群集，如果一个节点更新了设备的状态，他是否会立即将此更新通知所有其他节点？如果否，则mapWithState需要设置集群功能。我该怎么办呢？

我开发了一个 python (v3.6) 项目，该项目在我的AWS 生产平台的无服务器 lambda 中部署并实际工作。该项目使用以下依赖项：

- awscli==1.16.10
- boto3==1.9.0
- botocore==1.12.0
- psycopg2==2.7.5
- SQLAlchemy==1.2.11
- SQLAlchemy-Utils==0.33.3
- jsonschema==2.6.0

现在我想使用 localstack 在一个 lambda 中部署这个项目，我可以调用它来测试我的调用代码（不是我的项目实际工作的代码！）

要部署项目，首先我安装依赖项并创建一个 zip 文件：

pip install -r requirements.txt --upgrade -t ./
chmod -R 755 .
zip -r lambda.zip .

然后，我使用命令（使用正在运行的 localstack）：

aws lambda --region eu-east-1 --endpoint localhost:4574 \
    create-function --function-name mylambda \
    --zip-file fileb://lambda.zip

但是，它从 localstack 返回错误：

....
localstack_1  |     from functools32 import lru_cache
localstack_1  | ImportError: No module named functools32
....
localstack_1 …