我有一个由spark-ec2脚本设置的aws ec2集群.

我想配置Zeppelin,以便我可以在Zeppelin上本地编写scala代码并在集群上运行(通过master).此外,我希望能够访问我的s3桶.

我按照本指南和另一个,但我似乎无法从zeppelin运行scala代码到我的集群.

我在本地安装了Zeppelin

mvn install -DskipTests -Dspark.version=1.4.1 -Dhadoop.version=2.7.1

我的安全组设置为AmazonEC2FullAccess和AmazonS3FullAccess.

我在Zeppelin Webapp上编辑了spark解释器属性,火花://.us-west-2.compute.amazonaws.com:7077 来自本地[*]

1. 我测试的时候

   sc
   

   Run Code Online (Sandbox Code Playgroud)

   在翻译中,我收到了这个错误

   java.net.ConnectException: Connection refused at java.net.PlainSocketImpl.socketConnect(Native Method) at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:345) at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206) at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188) at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392) at java.net.Socket.connect(Socket.java:589) at org.apache.thrift.transport.TSocket.open(TSocket.java:182) at 
   

   Run Code Online (Sandbox Code Playgroud)

2. 当我尝试编辑"conf/zeppelin-site.xml"将我的端口更改为8082时,没有区别.

注意:我最终还想要访问我的s3存储桶,例如:

sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", "xxx")
sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey","xxx")
val file = "s3n://<<bucket>>/<<file>>"
val data = sc.textFile(file)
data.first

如果有任何仁慈的用户有任何建议(尚未在StackOverflow上发布),请告诉我!

它可以与火花纱客户一起很好地运行，但是纱线簇又如何呢？如果可能的话，请分享我们如何做到这一点。

我的配置如下:

Ubuntu 15.04
Java 1.7
Spark 1.4.1
Hadoop 2.7
Maven 3.3.3

我试图在从github成功克隆它并使用以下命令后安装Apache Zeppelin

mvn clean package -DskipTests

尽管进行了多次尝试,但在初步成功后,我收到以下错误:

[错误]无法在项目zeppelin-web上执行目标com.github.eirslett:frontend-maven-plugin:0.0.23:grunt(grunt build):无法运行任务:'grunt --no-color'失败.(错误代码3) - > [帮助1]

任何帮助将受到高度赞赏.提前致谢

问候,

Naveen Xavier

在Zeppelin,每行我都要提供每行的翻译.有没有办法为整个会话设置解释器.

%pyspark 

import re

花了0秒.

import pandas as pd

控制台:1:错误:'.' 预期但找到了标识符.将pandas导入为pd

%pyspark

import pandas as pd

花了0秒.

如何为整个会话设置解释器？

是否可以在AWS EMR 4.3上将Presto解释器添加到Zeppelin，如果可以，有人可以发布说明吗？我在EMR上运行了Presto-Sandbox和Zeppelin-Sandbox。

1. 我们如何从命令行运行笔记本？

2. 除了1，我如何将命令行参数传递到笔记本中？从笔记本代码中访问命令行参数？

是否可以注册用Scala编写的UDF(或函数)在PySpark中使用？例如:

val mytable = sc.parallelize(1 to 2).toDF("spam")
mytable.registerTempTable("mytable")
def addOne(m: Integer): Integer = m + 1
// Spam: 1, 2

在Scala中,现在可以使用以下内容:

val UDFaddOne = sqlContext.udf.register("UDFaddOne", addOne _)
val mybiggertable = mytable.withColumn("moreSpam", UDFaddOne(mytable("spam")))
// Spam: 1, 2
// moreSpam: 2, 3

我想在PySpark中使用"UDFaddOne"

%pyspark

mytable = sqlContext.table("mytable")
UDFaddOne = sqlContext.udf("UDFaddOne") # does not work
mybiggertable = mytable.withColumn("+1", UDFaddOne(mytable("spam"))) # does not work

背景:我们是一个开发人员团队,一些用Scala编写,一些用Python编写,并且想分享已编写的函数.也可以将其保存到库中并导入它.

我在Zeppelin的Python段落中创建了一个spark DataFrame.

sqlCtx = SQLContext(sc)
spDf = sqlCtx.createDataFrame(df)

并且df是一个熊猫数据帧

print(type(df))
<class 'pandas.core.frame.DataFrame'>

我想要做的是spDf从一个Python段落转到另一个Scala段落.看起来合理的方法是使用z.put.

z.put("spDf", spDf)

我收到了这个错误:

AttributeError: 'DataFrame' object has no attribute '_get_object_id'

有任何修复错误的建议吗？或任何移动的建议spDf？

我有一个常规的sbt scala spark项目，并希望与Zeppelin集成。可以想象，该项目不仅包含单个大文件，而且还包含一些用于模块化功能的不错的类。

我还不清楚如何将这两个世界融为一体，从而最好地转化为齐柏林飞艇。现在，我只能将过去的所有代码复制到一个齐柏林飞艇笔记本中，这并不是我真正想要实现的。

编辑

当在单个笔记本中编写代码并且所有代码都适合该笔记本时，齐柏林飞艇效果很好。但是，当代码变得更复杂时，我想将模块（可重用）封装为类（在单独的文件中）。我可以简单地在齐柏林飞艇中打开一个方向，然后使常规的Scala导入语句正常工作（例如在Jupiter中使用python）吗？

.
??? README.md
??? build.sbt
??? project
?   ??? build.properties
?   ??? plugins.sbt
??? scalastyle-config.xml
??? src
?   ??? main
?   ?   ??? resources
?   ?   ?   ??? log4j.properties
?   ?   ?   ??? zcta510.csv
?   ?   ??? scala
?   ?       ??? myOrg
?   ?           ??? MainApplication.scala
?   ?           ??? moduleAReferencedinMain.scala
?   ?           ??? bigModuleBAllSubPartsReferencedInMainViaImportStatements
?   ?               ??? B_1.scala
?   ?               ??? B_2.scala
?   ?               ??? B_3.scala
?   ??? test
?       ??? resources …

我正在（来自zeppelin 0.7的spark 2.1.0）中运行结构化流处理来自kafka的数据，并且试图通过spark.sql可视化流处理结果

如下 ：

%spark2
val spark = SparkSession
  .builder()
  .appName("Spark structured streaming Kafka example")
  .master("yarn")
  .getOrCreate()
val inputstream = spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "n11.hdp.com:6667,n12.hdp.com:6667,n13.hdp.com:6667 ,n10.hdp.com:6667, n9.hdp.com:6667")
    .option("subscribe", "st")
    .load()


val stream = inputstream.selectExpr("CAST( value AS STRING)").as[(String)].select(
             expr("(split(value, ','))[0]").cast("string").as("pre_post_paid"),
             expr("(split(value, ','))[1]").cast("double").as("DataUpload"),
             expr("(split(value, ','))[2]").cast("double").as("DataDowndownload"))
           .filter("DataUpload is not null and DataDowndownload is not null")
          .groupBy("pre_post_paid").agg(sum("DataUpload") + sum("DataDowndownload") as "size")
val query = stream.writeStream
.format("memory")
.outputMode("complete")
.queryName("test")
.start()

它运行后，我在“测试”上查询如下：

%sql
select *
from test

它仅在手动运行时更新，我的问题是如何处理新数据（流化可视化）时进行更新，如以下示例所示：

没有折中的见解：在Apache Spark中使用结构化流