小编ste*_*ino的帖子

我有两个独立的Kubernetes集群,分别用于登台和生产环境.我希望我将向每个集群提交Kubernetes API对象的YAML清单,但是某些对象在两个环境之间的配置会略有不同.

想象一下,在每个集群上运行一个内部Docker注册表,一个使用S3作为存储后端,一个使用GCS后端.注册表容器可以接受这些配置值作为环境变量或从文件中读取,这两个Kubernetes都表示支持,但是我应该如何为每个环境填充这些值？

基本上我想要的是一种让清单看起来像这样的方法,其中$()语法是在提交清单时在服务器上发生的变量插值:

---
apiVersion: v1
kind: Pod
metadata:
  name: foo
spec:
  containers:
    - name: foo
      image: foo
      env:
        - name: bar
          value: $(etcdctl get /path/to/bar)

我可以编写使用占位符作为值的模板,然后处理模板,从一些外部源中提取实际值,以生成实际提交给集群的最终清单.但是,我想我先问一个这样做的工具已经存在,或者有一些我不知道的Kubernetes祝福方式.这似乎是许多人为了使用Kubernetes而需要做的事情.

我曾经有过的其他想法包括在主机节点上使用etcd,confd和scripts的某些组合,但是这开始进入主机配置管理领域,我想以几乎所有成本避免这些领域.我正在使用CoreOS,主机完全通过coreos-cloudinit进行配置.换句话说,在创建节点时未定义的主机系统上没有任何操作,因此像Ansible这样的传统配置管理工具已经完成.

我知道Kubernetes的秘密系统,但其中一些变量值可能非常大,而且大多数都不是秘密.

我正在考虑使用Apache Kafka作为事件存储来在微服务中存储事件.

我通过各种博客阅读的一件事是,Kafka可以被认为是单一的事实来源,其中Kafka日志将存储给定主题的所有事件.

我想知道Kafka是否有能力从一开始就重播消息(例如,如果发生硬盘驱动器/网络崩溃)？

(请注意,我发现在主题目录下的/ tmp文件夹中存储了一些日志).有没有人知道可以调用以重播主题中的消息的任何命令(如果有的话)？

我有一个很大的数据集，我想删除包含null值的列并返回一个新的数据框。我怎样才能做到这一点？

以下仅删除包含null.

df.where(col("dt_mvmt").isNull()) #doesnt work because I do not have all the columns names or for 1000's of columns
df.filter(df.dt_mvmt.isNotNull()) #same reason as above
df.na.drop() #drops rows that contain null, instead of columns that contain null

例如

a |  b  | c
1 |     | 0
2 |  2  | 3

在上述情况下，它将删除整列，B因为其中一个值为空。

我有两个实体,如A和B,这样:

 class B {
    private Integer id;
    private String field1;
    private String field2;
    // getters and setters   
 }

 class A {
    private Date date;
    //one to one mapping is there between A and B
    private B b;
    //getters and setters
 }

我有一个spring数据存储库,以便:

 @Query("from A a where a.date= :date and a.b.id =:#{#b.id}")
        A findByBAndDate(@Param(value = "date") Date date,@Param(value = "b") B b);

但我得到了例外,no parameter binding found for name b!.

但是,如果我将上述查询修改为:

@Query("from A a where a.b.id =:#{#b.id}")
A findByB(@Param(value = …

我正面临与 Kafka 相关的问题。

我的当前服务 ( Producer) 将消息发送到 Kafka 主题 ( events)。该服务正在使用kafka_2.12 v1.0.0，用 Java 编写。

我正在尝试将它与spark-streaming作为Consumer服务的示例项目集成（这里使用 kafka_2.11 v0.10.0，用 Scala 编写）

消息成功发送Producer到 Kafka 主题。但是，我总是收到以下错误堆栈：

Exception in thread "main" org.apache.kafka.common.errors.InconsistentGroupProtocolException: The group member's supported protocols are incompatible with those of existing members.
    at ... run in separate thread using org.apache.spark.util.ThreadUtils ... ()
    at org.apache.spark.streaming.StreamingContext.liftedTree1$1(StreamingContext.scala:577)
    at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:571)    at com.jj.streaming.ItemApp$.delayedEndpoint$com$jj$streaming$ItemApp$1(ItemApp.scala:72)
    at com.jj.streaming.ItemApp$delayedInit$body.apply(ItemApp.scala:12)
    at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
    at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)     at scala.App$$anonfun$main$1.apply(App.scala:76)
    at scala.App$$anonfun$main$1.apply(App.scala:76)
    at scala.collection.immutable.List.foreach(List.scala:381)
    at …

在Akka中，您可以在构造函数中初始化actor的状态。此外，作为actor生命周期的一部分，preStart可以重写此方法以初始化actor的状态，并将其称为：

• 角色首次启动时，开始处理其第一条消息之前
• 当actor重新启动时，除非 postRestart被覆盖以不调用它

两种方式（构造函数和preStart）似乎覆盖了参与者启动阶段的重叠阶段，但有一点细微的差别，即保证可以在重新启动时调用构造函数preStart，但是只要您重写，就可以跳过构造函数postRestart。

在这种情况下，是否存在关于使用哪种最佳实践或模式的公认或书面记录？

此处讨论了扩大联合类型，但我似乎无法找到以下案例的答案

让我们从以下内容开始

val x = List(1, 2, "a")

这个异构列表被推断为List[Any]就像在 Scala 2 中一样

然而以下

val x2 = List(List(1, 2), Vector("a", "b"))

推断为 List[scala.collection.immutable.AbstractSeq[Int | String]]

这是相当混乱的行为。为什么Any在一种情况下会推断出两种不相交类型的 LUB ，而在另一种情况下会推断出联合类型？

如果这只是一个设计决定，是否有任何此类情况需要人们注意？

当我试图用seaborn保存我的情节时,像这样:

import seaborn as sn
import pandas as pd
import matplotlib.pyplot as plt
from pylab import savefig

array = [[100,0], 
        [33,67]]

df_cm = pd.DataFrame(array)

svm = sn.heatmap(df_cm, annot=True,cmap='coolwarm', linecolor='white', linewidths=1)

svm.savefig('svm_conf.png', dpi=400)

我收到这个错误

AttributeError                            Traceback (most recent call last)
<ipython-input-71-5c0ae9cda020> in <module>()
----> 1 svm.savefig('svm_conf.png', dpi=400)

AttributeError: 'AxesSubplot' object has no attribute 'savefig'

我以前用相同的代码保存了一些箱图,但这一次,它不起作用.

我有一个Spark应用程序,现在运行了46个小时.虽然大多数工作在25秒内完成,但特定工作需要数小时.下面提供了一些细节:

Task Time   Shuffle Read        Shuffle Write
7.5 h       2.2 MB / 257402     2.9 MB / 128601

还有其他类似的任务时间,其中值为11.3小时,10.6小时,9.4小时等.每个小时的活动时间花费在"数据框架上的rdd:www.cacala:42.".阶段的细节显示执行者花费在"执行者计算时间"上的时间.此执行程序在DataNode 1上运行,其中CPU利用率非常正常,约为13%.其他盒子(另外4个工作节点)具有非常标称的CPU利用率.

当Shuffle Read在5000条记录内时,这非常快,并且如前所述在25秒内完成.没有任何内容附加到日志(spark/hadoop/hbase),也没有注意到任何位置/tmp或/var/tmp位置,这表明某些磁盘相关活动正在进行中.

我对于出了什么问题一无所知.很长一段时间以来一直在努力解决这个问题.使用的软件版本如下:

Hadoop    : 2.7.2
Zookeeper : 3.4.9
Kafka     : 2.11-0.10.1.1
Spark     : 2.1.0
HBase     : 1.2.6
Phoenix   : 4.10.0

spark默认文件中的一些配置.

spark.eventLog.enabled           true
spark.eventLog.dir               hdfs://SDCHDPMAST1:8111/data1/spark-event
spark.history.fs.logDirectory    hdfs://SDCHDPMAST1:8111/data1/spark-event
spark.yarn.jars                  hdfs://SDCHDPMAST1:8111/user/appuser/spark/share/lib/*.jar
spark.driver.maxResultSize       5G
spark.deploy.zookeeper.url       SDCZKPSRV01

spark.executor.memory                   12G
spark.driver.memory                     10G
spark.executor.heartbeatInterval        60s
spark.network.timeout                   300s

有什么办法可以减少"执行计算时间"所花费的时间吗？

我目前正在将之前在bash脚本中实现的工作流转换为Airflow DAG。在bash脚本中，我只是在运行时使用以下命令导出变量

export HADOOP_CONF_DIR="/etc/hadoop/conf"

现在我想在Airflow中做同样的事情，但是还没有找到解决方案。我发现的一种解决方法是在os.environ[VAR_NAME]='some_text'任何方法或运算符外部设置变量，但这意味着在脚本加载后而不是在运行时将它们导出。

现在，当我尝试调用os.environ[VAR_NAME] = 'some_text'由PythonOperator 调用的函数时，它不起作用。我的代码看起来像这样

def set_env():
    os.environ['HADOOP_CONF_DIR'] = "/etc/hadoop/conf"
    os.environ['PATH'] = "somePath:" + os.environ['PATH']
    os.environ['SPARK_HOME'] = "pathToSparkHome"
    os.environ['PYTHONPATH'] = "somePythonPath"
    os.environ['PYSPARK_PYTHON'] = os.popen('which python').read().strip()
    os.environ['PYSPARK_DRIVER_PYTHON'] = os.popen('which python').read().strip()

set_env_operator = PythonOperator(
    task_id='set_env_vars_NOT_WORKING',
    python_callable=set_env,
    dag=dag)

现在，当我的SparkSubmitOperator被执行时，我得到了异常：

Exception in thread "main" java.lang.Exception: When running with master 'yarn' either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment.

我使用的情况下，这是相关的是，我有SparkSubmitOperator，我在那里作业提交到纱，因此无论是HADOOP_CONF_DIR或YARN_CONF_DIR必须在环境中进行设置。.bashrc对我来说，在我或其他任何配置中设置它们都是遗憾的，这就是为什么我需要在运行时设置它们。

最好在执行之前SparkSubmitOperator，先在操作员中设置它们，但如果有可能将它们作为参数传递给SparkSubmitOperator，那至少是可以的。