小编Tim*_*m B的帖子

我正在尝试找到一种方法来为swig生成的函数添加代码.我使用了类型映射来扩展类,但在文档中找不到有关扩展特定函数的任何内容.

给出以下swig接口文件:

%module Test
%{
#include "example.h"
%}

%typemap(cscode) Example %{
    bool 64bit = SizeOf(typeof(System.IntPtr)) == 8;
    static string Path = 64bit ? "/...Path to 64 bit dll.../" : 
                                 "/...Path to 32 bit dll.../";
%}

%include "example.h"

我得到以下C#代码:

public class MyClass : global::System.IDisposable {
    ...
    bool 64bit = SizeOf(typeof(System.IntPtr)) == 8;
    static string Path = 64bit ? "/...Path to 64 bit dll.../" : 
                                 "/...Path to 32 bit dll.../";

    ...
    public static SomeObject Process(...) {     // Function defined in example.h
                                               <- …

我是pyspark的新手,我正在尝试使用它处理一个大型数据集,该数据集保存为csv文件.我想将CSV文件读入spark数据帧,删除一些列,然后添加新列.我该怎么办？

我无法将此数据转换为数据帧.这是我到目前为止的简化版本:

def make_dataframe(data_portion, schema, sql):
    fields = data_portion.split(",")
    return sql.createDateFrame([(fields[0], fields[1])], schema=schema)

if __name__ == "__main__":
    sc = SparkContext(appName="Test")
    sql = SQLContext(sc)

    ...

    big_frame = data.flatMap(lambda line: make_dataframe(line, schema, sql))
                .reduce(lambda a, b: a.union(b))

    big_frame.write \
        .format("com.databricks.spark.redshift") \
        .option("url", "jdbc:redshift://<...>") \
        .option("dbtable", "my_table_copy") \
        .option("tempdir", "s3n://path/for/temp/data") \
        .mode("append") \
        .save()

    sc.stop()

这会TypeError: 'JavaPackage' object is not callable在reduce步骤中产生错误.

是否有可能做到这一点？减少到数据帧的想法是能够将结果数据写入数据库(Redshift,使用spark-redshift包).

我一直在使用也试过unionAll(),而且map()用partial(),但不能让它开始工作.

我在亚马逊的EMR上运行它spark-redshift_2.10:2.0.0,并使用亚马逊的JDBC驱动程序RedshiftJDBC41-1.1.17.1017.jar.

我正在寻找一种方法来运行spark.ml.feature.PCA函数对数据groupBy()帧上的调用返回的分组数据.但我不确定这是否可能,或者如何实现它.这是一个基本的例子,希望能说明我想做的事情:

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.feature import PCA   

df = spark.createDataFrame([[3, 1, 1], [4, 2, 1], [5, 2, 1], [3, 3, 2], [6, 2, 2], [4, 4, 2]], ["Value1", "Value2",  "ID"])

df.show()
+------+------+---+
|Value1|Value2| ID|
+------+------+---+
|     3|     1|  1|
|     4|     2|  1|
|     5|     2|  1|
|     3|     3|  2|
|     6|     2|  2|
|     4|     4|  2|
+------+------+---+

assembler = VectorAssembler(inputCols=["Value1", "Value2"], outputCol="features")

df2 = assembler.transform(df)

df2.show()
+------+------+---+---------+
|Value1|Value2| ID| features|
+------+------+---+---------+ …

有没有办法在EMR集群的所有节点上设置环境变量？

我在Python3 PySpark中尝试使用reduceByKey()时遇到错误,并且收到有关散列种子的错误.我可以看到这是一个已知的错误,并且环境变量PYTHONHASHSEED需要在集群的所有节点上设置为相同的值,但我没有任何运气.

我尝试通过群集配置向spark-env添加变量:

[
  {
    "Classification": "spark-env",

      "Configurations": [
      {
         "Classification": "export",
         "Properties": {
          "PYSPARK_PYTHON": "/usr/bin/python3",
          "PYTHONHASHSEED": "123"
       }
     }
   ]
 },
 {
   "Classification": "spark",
   "Properties": {
     "maximizeResourceAllocation": "true"
    }
  }
]

但这不起作用.我还尝试添加一个引导脚本:

#!/bin/bash
export PYTHONHASHSEED=123

但这似乎也没有做到这一点.