小编Gya*_*yan的帖子

我最近开始了一个PoC项目,我们正在开发一个小型的Web应用程序.初始设置在AWS的微实例上完成.我们在rails + mysql堆栈上.

在安装/运行MySQL之后,我发现已经消耗了大约500多MB的RAM; 其余的系统留下的相当少(微型实例只有620 MB RAM).

我们的应用程序在此阶段相当简单.我可以做些什么来减少MySQL服务器消耗的内存吗？

感谢帮助.

最近,在处理db2 - > oracle迁移项目时,我们遇到了这种情况.开发人员无意中使用decimal(s,p)列创建新的表结构.我不记得Oracle支持这个,但后来一些挖掘表明它是一个ANSI数据类型,因此由oracle支持.

但问题仍然存在 -

1. 这些数据是如何在内部处理的？
2. 是否需要使用ANSI类型而不是Oracle的内置类型？
3. 如果目标类型是Oracle内置类型,那么在数据迁移期间是否会产生影响？

在Ubuntu上,我正在尝试使用wget下载文件(从脚本中).构建程序以每天下载此文件并加载到hadoop集群.

但是,wget失败,并显示以下消息.

wget http://www.nseindia.com/content/historical/EQUITIES/2012/JUN/cm15JUN2012bhav.csv.zip
--2012-06-16 03:37:30--  http://www.nseindia.com/content/historical/EQUITIES/2012/JUN/cm15JUN2012bhav.csv.zip
Resolving www.nseindia.com... 122.178.225.48, 122.178.225.18
Connecting to www.nseindia.com|122.178.225.48|:80... connected.
HTTP request sent, awaiting response... 403 Forbidden
2012-06-16 03:37:30 ERROR 403: Forbidden.

当我在firefox或同等版本中尝试相同的url时,它工作得很好.是的,没有涉及许可协议的事情......

我错过了关于wget的基本内容吗？

谷歌地图的新手,我最近开始阅读.

对于其中一个项目,我不得不在地图上绘制几个点.

我从Google的API主页获取了一个演示/样本片段用于地图,并开始根据我的要求进行调整.一切顺利.

但是,在某个时间点,我决定插入一个按钮,需要单击该按钮以显示地图,否则默认情况下不会显示地图.

仍然,页面加载没有错误,但是,一旦我单击按钮,我收到此消息" 资源解释为图像但传输为text/html "

尝试弄清楚我是否在某处明确地设置了内容类型但没有找到任何内容.

有人见过吗？会很感激帮助.

我基本上是从Kafka源代码中读取,并将每条消息转发到我的foreach处理器(感谢Jacek的简单示例页面).

如果这确实有效,我将在process这里实际执行一些业务逻辑,但是,这不起作用.我相信println自从它在执行程序上运行后它不起作用,并且没有办法将这些日志恢复到驱动程序.但是,这个insert into临时表至少应该起作用,并告诉我消息实际上被消耗并处理到接收器.

我在这里错过了什么？

真的在寻找第二组眼睛来检查我的努力:

 val stream = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", Util.getProperty("kafka10.broker")) 
      .option("subscribe", src_topic) 
      .load()

    val rec = stream.selectExpr("CAST(value AS STRING) as txnJson").as[(String)]

    val df = stream.selectExpr("cast (value as string) as json")

    val writer = new ForeachWriter[Row] {
      val scon = new SConConnection
      override def open(partitionId: Long, version: Long) = {
        true
      }
      override def process(value: Row) = {
        println("++++++++++++++++++++++++++++++++++++" + value.get(0))
        scon.executeUpdate("insert into rs_kafka10(miscCol) values("+value.get(0)+")")
      }
      override def close(errorOrNull: Throwable) …

我正在向YARN(在spark 2.1.1 + kafka 0.10.2.1上)提交一份工作,该工作连接到一个安全的hbase集群.当我在"本地"模式下运行时,这项工作表现得很好(spark.master = local [*]).

但是,只要我将master作为YARN提交作业(并将部署模式作为客户端),我就会看到以下错误消息 -

Caused by: javax.security.auth.login.LoginException: Unable to obtain password from user

我正在按照hortonworks的建议向纱线集群提供关于HBase和keytab等的信息.遵循这篇kb文章 - https://community.hortonworks.com/content/supportkb/48988/how-to-run-spark-job-to -interact与-固定-hbas.html

什么指针可能会发生什么？

登录HBase的机制:

UserGroupInformation.setConfiguration(hbaseConf)
val keyTab = "keytab-location") 
val principal = "kerberos-principal"
val ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keyTab)
UserGroupInformation.setLoginUser(ugi)
ugi.doAs(new PrivilegedExceptionAction[Void]() {

override def run: Void = {
  hbaseCon = Some(ConnectionFactory.createConnection(hbaseConf))
  null
}
})

此外,我尝试了登录的替代机制,如下所示:

UserGroupInformation.loginUserFromKeytab(principal, keyTab)
connection=ConnectionFactory.createConnection(hbaseConf)

请建议.

作为R环境的真正新手,我将阅读"R简介"一书.

在一个例子中,作者建议"

表达式中较短的向量经常根据需要(可能是分数)再循环,直到它们与最长向量的长度匹配.

紧接着,有这个例子......其中它表明一个向量将重复2.2次......

但是,当我在我的系统上复制相同的示例(ubuntu 64b,R - v2.4.11)时,我收到此错误消息

x
[1]  2  5  8  6 11
> y
[1] 23 11
> v=2*x+y+1
Warning message:
In 2 * x + y :
  longer object length is not a multiple of shorter object length
> v
[1] 28 22 40 24 46

试图在内部搜索谷歌,stackoverflow,但无法找到任何令人满意的...我在这里错过了什么？是否有使用RI版本的东西？

我正在使用Spark 2.1.1(kafka 0.10+)阅读Kafka主题,有效负载是JSON字符串.我想用模式解析字符串并继续使用业务逻辑.

每个人似乎都建议我应该使用from_json解析JSON字符串,但是,它似乎不适合我的情况编译.错误是

not found : value from_json 
.select(from_json($"json", txnSchema) as "data")

当我尝试将以下几行放入火花壳时,它的效果很好 -

val df = stream
  .select($"value" cast "string" as "json")
  .select(from_json($"json", txnSchema) as "data")
  .select("data.*")

任何想法,我在代码中可能做错了,看到这个部分在shell中运行但在IDE /编译时没有？

这是代码:

import org.apache.spark.sql._

object Kafka10Cons3 extends App {
  val spark = SparkSession
    .builder
    .appName(Util.getProperty("AppName"))
    .master(Util.getProperty("spark.master"))
    .getOrCreate

  val stream = spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", Util.getProperty("kafka10.broker"))
    .option("subscribe", src_topic)
    .load

  val txnSchema = Util.getTxnStructure
  val df = stream
    .select($"value" cast "string" as "json")
    .select(from_json($"json", txnSchema) as "data")
    .select("data.*")
}

我想使用MergeContent处理器将推文合并到Elasticsearch索引中.为此我需要命令和推文(每个)由\n分隔.它应该是这样的

{ action: { metadata }}\n
{ request body        }\n

这将是

{ “index”}
{ tweet1 }
{ tweet2 } 

当我将\n作为分隔符时,处理器实际上将\n添加为字符串而不是新行分隔符.有可能使它成为真正的新线吗？也可以将页脚留空？

提前致谢.

我们使用的是HDP 2.4.2,使用Scala 2.10.5编译spark 1.6.Hbase版本为1.1.2.2.4.2.0-258

环境是一个基本的开发集群(<10个节点),其中hbase和spark在集群模式下运行.

尝试使用spark hbase连接器将hbase中的soem数据导入spark中的数据帧失败,出现以下错误 -

Exception in thread "main" java.lang.UnsupportedOperationException: empty.tail
    at scala.collection.TraversableLike$class.tail(TraversableLike.scala:445)
    at scala.collection.mutable.ArraySeq.scala$collection$IndexedSeqOptimized$super$tail(ArraySeq.scala:45)
    at scala.collection.IndexedSeqOptimized$class.tail(IndexedSeqOptimized.scala:123)
    at scala.collection.mutable.ArraySeq.tail(ArraySeq.scala:45)
    at org.apache.spark.sql.execution.datasources.hbase.HBaseTableCatalog.initRowKey(HBaseTableCatalog.scala:150)
    at org.apache.spark.sql.execution.datasources.hbase.HBaseTableCatalog.<init>(HBaseTableCatalog.scala:164)
    at org.apache.spark.sql.execution.datasources.hbase.HBaseTableCatalog$.apply(HBaseTableCatalog.scala:239)
    at hbaseReaderHDPCon$.main(hbaseReaderHDPCon.scala:42)
    at hbaseReaderHDPCon.main(hbaseReaderHDPCon.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:497)
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$runMain(SparkSubmit.scala:731)
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181)
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

在我的代码的第42行 - 这发生了 -

val cat =
      s"""{
          |"table":{"namespace":"myTable", "name":"person", "tableCoder":"PrimitiveType"},
          |"rowkey":"ROW",
          |"columns":{
            |"col0":{"cf":"person", "col":"detail", "type":"string"}
          |}
          |}""".stripMargin
    val scon = new SparkConf()
    val sparkContext = new SparkContext(scon)

我正在Spark 2.1.1中使用结构化流。我需要对传入消息（从Kafka来源）应用一些业务逻辑。

本质上，我需要获取消息，获取一些键值，在HBase中查找它们并在数据集上执行更多biz逻辑。最终结果是一个字符串消息，需要将其写出到另一个Kafka队列。

但是，由于传入消息的抽象是一个数据帧（无边界表-结构化流），因此我必须遍历一次触发期间接收到的数据集mapPartitions（由于HBase客户端无法序列化而导致的分区）。

在我的流程中，我需要遍历每一行以执行相同的业务流程。

1. 有没有更好的方法可以帮助我避免dataFrame.mapPartitions打电话呢？我觉得它的顺序和迭代！
2. 结构化流基本上迫使我从业务流程中生成输出数据帧，而没有一开始。我可以使用其他哪些设计模式来实现最终目标？

您会推荐一种替代方法吗？