小编Ath*_*kur的帖子

我正在使用 Zeppelin 运行 Scala Spark 作业。当我运行它时，我收到以下错误：

latestForEachKey: org.apache.spark.sql.DataFrame = [PartitionStatement_1: string, PartitionYear_1: string ... 64 more fields]
<console>:393: error: Could not write class $$$$2e6199f161363585e7ae9b28bcf8535e$$$$iw because it exceeds JVM code size limits. Method <init>'s code too large!
class $iw extends Serializable {

有时我没有收到错误并且它有效。我能做些什么来解决这个问题？

这是我运行的代码：

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._

import org.apache.spark.{ SparkConf, SparkContext }
import java.sql.{Date, Timestamp}
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions.udf
import java.io.File
import org.apache.hadoop.fs._
import org.apache.spark.sql.functions.input_file_name
import org.apache.spark.sql.functions.regexp_extract

val getPartition = spark.udf.register("getPartition", (filePath: String) => filePath.split("\\.")(3)) …

我有两个数据框，正在对 5 列执行外连接。下面是我的数据集的示例。

uniqueFundamentalSet|^|PeriodId|^|SourceId|^|StatementTypeCode|^|StatementCurrencyId|^|FinancialStatementLineItem.lineItemId|^|FinancialAsReportedLineItemName|^|FinancialAsReportedLineItemName.languageId|^|FinancialStatementLineItemValue|^|AdjustedForCorporateActionValue|^|ReportedCurrencyId|^|IsAsReportedCurrencySetManually|^|Unit|^|IsTotal|^|StatementSectionCode|^|DimentionalLineItemId|^|IsDerived|^|EstimateMethodCode|^|EstimateMethodNote|^|EstimateMethodNote.languageId|^|FinancialLineItemSource|^|IsCombinedItem|^|IsExcludedFromStandardization|^|DocByteOffset|^|DocByteLength|^|BookMark|^|ItemDisplayedNegativeFlag|^|ItemScalingFactor|^|ItemDisplayedValue|^|ReportedValue|^|EditedDescription|^|EditedDescription.languageId|^|ReportedDescription|^|ReportedDescription.languageId|^|AsReportedInstanceSequence|^|PhysicalMeasureId|^|FinancialStatementLineItemSequence|^|SystemDerivedTypeCode|^|AsReportedExchangeRate|^|AsReportedExchangeRateSourceCurrencyId|^|ThirdPartySourceCode|^|FinancialStatementLineItemValueUpperRange|^|FinancialStatementLineItemLocalLanguageLabel|^|FinancialStatementLineItemLocalLanguageLabel.languageId|^|IsFinal|^|FinancialStatementLineItem.lineItemInstanceKey|^|StatementSectionIsCredit|^|CapitalChangeAdjustmentDate|^|ParentLineItemId|^|EstimateMethodId|^|StatementSectionId|^|SystemDerivedTypeCodeId|^|UnitEnumerationId|^|FiscalYear|^|IsAnnual|^|PeriodPermId|^|PeriodPermId.objectTypeId|^|PeriodPermId.objectType|^|AuditID|^|AsReportedItemId|^|ExpressionInstanceId|^|ExpressionText|^|FFAction|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|221|^|Average Age of Employees|^|505074|^|30.00000|^||^||^|False|^|1.00000|^|False|^|EMP|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|0|^||^||^||^|505074|^||^|505074|^||^||^|122880|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235002211206722736|^|True|^||^||^|3019656|^|3013652|^|3019679|^|1010066|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|498|^|Shareholders' Equity Per Share|^|505074|^|91.37000|^|678.74654|^|500186|^|False|^|1.00000|^|False|^|TAN|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|0|^||^||^||^|505074|^||^|505074|^||^||^|474880|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235004981302988315|^|True|^||^||^|3019656|^|3013751|^|3019679|^|1010066|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|500|^|Number of Shares Outstanding at Period End-Common Shares|^|505074|^|90000000.00000|^|12115420.96161|^||^|False|^|1000.00000|^|False|^|TAN|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|3|^||^||^||^|505074|^||^|505074|^||^||^|499712|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235005001178855709|^|True|^||^||^|3019656|^|3013751|^|3019679|^|1010067|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|562|^|Number of Employees|^|505074|^|2924.00000|^||^||^|False|^|1.00000|^|False|^|EMP|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|0|^||^||^||^|505074|^||^|505074|^||^||^|464864|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235005621461877526|^|True|^||^||^|3019656|^|3013652|^|3019679|^|1010066|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|655|^|Total number of shareholders|^|505074|^|11792.00000|^||^||^|False|^|1.00000|^|False|^|OTH|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|0|^||^||^||^|505074|^||^|505074|^||^||^|466927|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235006551335570418|^|True|^||^||^|3019656|^|3013716|^|3019679|^|1010066|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|657|^|Total dividends paid (common stock)|^|505074|^|540000000.00000|^||^|500186|^|False|^|1000000.00000|^|False|^|OTH|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|6|^||^||^||^|505074|^||^|505074|^||^||^|233463|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|12350065712483219|^|True|^||^||^|3019656|^|3013716|^|3019679|^|1010068|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|1452|^|Order received|^|505074|^|26936000000.00000|^||^|500186|^|False|^|1000000.00000|^|False|^|OTH|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|6|^||^||^||^|505074|^||^|505074|^||^||^|350195|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235014521608462544|^|True|^||^||^|3019656|^|3013716|^|3019679|^|1010068|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|1453|^|Order backlogs|^|505074|^|1447000000.00000|^||^|500186|^|False|^|1000000.00000|^|False|^|OTH|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|6|^||^||^||^|505074|^||^|505074|^||^||^|350195|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235014531922884465|^|True|^||^||^|3019656|^|3013716|^|3019679|^|1010068|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|1457|^|Export amount|^|505074|^|3924000000.00000|^||^|500186|^|False|^|1000000.00000|^|False|^|OTH|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|6|^||^||^||^|505074|^||^|505074|^||^||^|291829|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235014571728332413|^|True|^||^||^|3019656|^|3013716|^|3019679|^|1010068|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239205|^|235|^|1|^|FTN|^|500186|^|1459|^|Capital expenditures (Note)|^|505074|^|659000000.00000|^||^|500186|^|False|^|1000000.00000|^|False|^|OTH|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|6|^||^||^||^|505074|^||^|505074|^||^||^|350195|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1235014591148256870|^|True|^||^||^|3019656|^|3013716|^|3019679|^|1010068|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239285|^|236|^|1|^|FTN|^|500186|^|255|^|Number of Employees|^|505074|^|10152.00000|^||^||^|False|^|1.00000|^|False|^|EMP|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|0|^||^||^||^|505074|^||^|505074|^||^||^|12288|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1236002551128894330|^|True|^||^||^|3019656|^|3013652|^|3019679|^|1010066|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239285|^|236|^|1|^|FTN|^|500186|^|256|^|Average Age of Employees|^|505074|^|34.00000|^||^||^|False|^|1.00000|^|False|^|EMP|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|0|^||^||^||^|505074|^||^|505074|^||^||^|122880|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1236002561111316467|^|True|^||^||^|3019656|^|3013652|^|3019679|^|1010066|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239285|^|236|^|1|^|FTN|^|500186|^|542|^|Shareholders' Equity Per Share|^|505074|^|160.20000|^|691.93184|^|500186|^|False|^|1.00000|^|False|^|TAN|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|0|^||^||^||^|505074|^||^|505074|^||^||^|471038|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1236005421170597389|^|True|^||^||^|3019656|^|3013751|^|3019679|^|1010066|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239285|^|236|^|1|^|FTN|^|500186|^|545|^|Number of Shares Outstanding at Period End-Common Shares|^|505074|^|679468000.00000|^|157314300.64243|^||^|False|^|1000.00000|^|False|^|TAN|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|3|^||^||^||^|505074|^||^|505074|^||^||^|472064|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1236005451445165969|^|True|^||^||^|3019656|^|3013751|^|3019679|^|1010067|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!|
192730239285|^|236|^|1|^|FTN|^|500186|^|718|^|Total dividends paid (common stock)|^|505074|^|4750000000.00000|^||^|500186|^|False|^|1000000.00000|^|False|^|OTH|^||^|False|^|ARV|^||^|505074|^||^|False|^|False|^||^||^||^||^|6|^||^||^||^|505074|^||^|505074|^||^||^|458752|^|NA|^||^||^|TK |^||^||^|505126|^|True|^|1236007181118043352|^|True|^||^||^|3019656|^|3013716|^|3019679|^|1010068|^|1976|^|True|^||^|1000220295|^||^||^||^||^||^|I|!| …

我有两个如下所示的数据框

+--------------------+--------+-----------+-------------+
|UniqueFundamentalSet|Taxonomy|FFAction|!||DataPartition|
+--------------------+--------+-----------+-------------+
|192730241374        |1       |I|!|       |Japan        |
|192730241374        |2       |I|!|       |Japan        |
|192730241373        |1       |I|!|       |Japan        |
|192730241373        |2       |I|!|       |Japan        |
+--------------------+--------+-----------+-------------+

+--------------------+--------+-----------+-------------+
|UniqueFundamentalSet|Taxonomy|FFAction|!||DataPartition|
+--------------------+--------+-----------+-------------+
|192730241374        |1       |I|!|       |Japan        |
|192730241374        |2       |I|!|       |Japan        |
|192730391384        |1       |I|!|       |Japan        |
|192730391384        |2       |I|!|       |Japan        |
|192730241373        |1       |I|!|       |Japan        |
|192730241373        |2       |I|!|       |Japan        |
+--------------------+--------+-----------+-------------+

当我在上述数据框之间执行联合时，我得到重复的行。这是我的输出

+--------------------+--------+-----------+-------------+
|UniqueFundamentalSet|Taxonomy|FFAction|!||DataPartition|
+--------------------+--------+-----------+-------------+
|192730241374        |1       |I|!|       |Japan        |
|192730241374        |2       |I|!|       |Japan        |
|192730241373        |1       |I|!|       |Japan        |
|192730241373        |2 …

我在 s3 文件夹中有 Spark 输出，我想将所有 s3 文件从该输出文件夹移动到另一个位置，但在移动时我想重命名这些文件。

例如，我在 S3 文件夹中有文件，如下所示

现在我想重命名所有文件并放入另一个目录中，但文件的名称如下所示

Fundamental.FinancialStatement.FinancialStatementLineItems.Japan.1971-BAL.1.2017-10-18-0439.Full.txt
Fundamental.FinancialStatement.FinancialStatementLineItems.Japan.1971-BAL.2.2017-10-18-0439.Full.txt
Fundamental.FinancialStatement.FinancialStatementLineItems.Japan.1971-BAL.3.2017-10-18-0439.Full.txt

这里 Fundamental.FinancialStatement 在所有文件的 2017-10-18-0439当前日期时间中都是常量。

这是我到目前为止所尝试过的，但无法获取文件夹名称并循环遍历所有文件

    import org.apache.hadoop.fs._

val src = new Path("s3://trfsmallfffile/Segments/output")
val dest = new Path("s3://trfsmallfffile/Segments/Finaloutput")
val conf = sc.hadoopConfiguration   // assuming sc = spark context
val fs = src.getFileSystem(conf)
//val file = fs.globStatus(new Path("src/DataPartition=Japan/part*.gz"))(0).getPath.getName
//println(file)
val status = fs.listStatus(src)    

status.foreach(filename => {
               val a = filename.getPath.getName.toString()
                println("file name"+a)
                //println(filename)
             })

这给了我以下输出

    file nameDataPartition=Japan
file nameDataPartition=SelfSourcedPrivate
file nameDataPartition=SelfSourcedPublic
file name_SUCCESS

这为我提供了文件夹详细信息，而不是文件夹内的文件。

参考资料取自这里Stack Overflow …

我试图读取Spark Scala中的大文件，然后尝试执行join。当我使用较小的文件进行测试时，效果很好，但是对于较大的文件，我得到了一些低于错误的时间。

我设法取出其中一个出错的文件。文件大小为1 GB，并且在最后创建分区时，在我分割文件名以获取列的地方抛出此错误。

就在这条线之后

 val rdd = sc.textFile(mainFileURL)
      val header = rdd.filter(_.contains("uniqueFundamentalSet")).map(line => line.split("\\|\\^\\|")).first()
      val schema = StructType(header.map(cols => StructField(cols.replace(".", "_"), StringType)).toSeq)
      println(schema)
      val data = sqlContext.createDataFrame(rdd.filter(!_.contains("uniqueFundamentalSet")).map(line => Row.fromSeq(line.split("\\|\\^\\|").toSeq)), schema)

This is the culprit line

 val data = sqlContext.createDataFrame(rdd.filter(!_.contains("uniqueFundamentalSet")).map(line => Row.fromSeq(line.split("\\|\\^\\|").toSeq)), schema)

Please suggest how can I handle this.

When I do rdd.count I get value . But when I do data.count() I get the error

Caused by: java.lang.RuntimeException: Error while encoding: java.lang.ArrayIndexOutOfBoundsException: 37
if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null …

我在S3中存储了大约一百万个文本文件。我想根据文件夹名称重命名所有文件。

我如何在Spark-Scala中做到这一点？

我正在寻找一些示例代码。

我正在使用齐柏林飞艇来运行我的spark脚本。

下面的代码我已经尝试从答案中建议

import org.apache.hadoop.fs._

val src = new Path("s3://trfsmallfffile/FinancialLineItem/MAIN")
val dest = new Path("s3://trfsmallfffile/FinancialLineItem/MAIN/dest")
val conf = sc.hadoopConfiguration   // assuming sc = spark context
val fs = Path.getFileSystem(conf)
fs.rename(src, dest)

但是低于错误

<console>:110: error: value getFileSystem is not a member of object org.apache.hadoop.fs.Path
       val fs = Path.getFileSystem(conf)

我有以下格式的数据框

+-----------------+-------------------------+-----------+-----------------------------+---------------------------+----------------------------------+--------------------------------------+--------------------------------+
|DataPartition    |TimeStamp                |FFAction|!||IdentifierValue_effectiveFrom|IdentifierValue_effectiveTo|IdentifierValue_identifierEntityId|IdentifierValue_identifierEntityTypeId|IdentifierValue_identifierTypeId|
+-----------------+-------------------------+-----------+-----------------------------+---------------------------+----------------------------------+--------------------------------------+--------------------------------+
|SelfSourcedPublic|2018-03-05T11:54:18+00:00|I|!|       |1900-01-01T00:00:00+00:00    |9999-12-31T00:00:00+00:00  |4295903126                        |404010                                |320150                          |
+-----------------+-------------------------+-----------+-----------------------------+---------------------------+----------------------------------+--------------------------------------+--------------------------------+

我想在下面的列中添加带有条件的额外列

IdentifierValue_identifierEntityTypeId

添加具有以下条件的额外列分区

如果 IdentifierValue_identifierEntityTypeId =1001371402 那么分区 =Repno2FundamentalSeries 否则如果 IdentifierValue_identifierEntityTypeId404010 那么分区 = Repno2Organization

这就是我正在努力实现的目标

 val temp = temp1.withColumn("Partition", when($"IdentifierValue_identifierEntityTypeId" === "404010", 0).otherwise("Repno2FundamentalSeries"))
    temp.show(false)

我的输出低于输出，但得到的值为零

IdentifierValue_identifierEntityTypeId

我是 Scala 的新手，所以提出了基本问题

对于列上的多个条件如何写 when 和 else 。这对我不起作用

线程“main”中的异常 java.lang.IllegalArgumentException：otherwise() 只能在先前由 when() 生成的列上应用一次

val dataMain = dataMain1.withColumn(
      "Partition",
      when($"RelationObjectId_relatedObjectType" === "EDInstrument" && $"RelationObjectId_relatedObjectType" === "Fundamental", "Instrument2Fundamental")
        .otherwise(when($"RelationObjectId_relatedObjectType" === "EDInstrument" && $"RelationObjectId_relatedObjectType" === "FundamentalSeries", "Instrument2FundamentalSeries"))
        .otherwise(when($"RelationObjectId_relatedObjectType" === "Organization" && $"RelationObjectId_relatedObjectType" === "Fundamental", …