1

我的文件大小约为 10 GB。我需要提取数据并将它们插入到多个配置单元表中。

我能够在单个节点中使用一些准备/映射功能来完成。

样本数据 ：

Dept : HR
Emp name is Andrew lives in Colorodo
DOB : 03/09/1958
Project name : Healthcare
DOJ : 06/04/2011
DOL : 09/21/2011
Project name : Retail
DOJ : 11/04/2011
DOL : 08/21/2013
Project name : Audit
DOJ : 09/11/2013
DOL : 09/01/2014
Project name : ContorlManagement
DOJ : 01/08/2015
DOL : 02/14/2016
Emp name is Alex lives in Texas
DOB : 03/09/1958
Project name : Healthcare
DOJ : 06/04/2011
DOL : 09/21/2011
Project name : ContorlManagement
DOJ : 01/08/2015
DOL : 02/14/2016
Emp name is Mathew lives in California
DOB : 03/09/1958
Project name : Healthcare
DOJ : 06/04/2011
DOL : 09/21/2011
Project name : Retail
DOJ : 11/04/2011
DOL : 08/21/2013
Project name : Audit
DOJ : 09/11/2013
DOL : 09/01/2014
Project name : ContorlManagement
DOJ : 01/08/2015
DOL : 02/14/2016

Dept : QC
Emp name is Nguyen lives in Nevada
DOB : 03/09/1958
Project name : Healthcare
DOJ : 06/04/2011
DOL : 09/21/2011
Project name : Retail
DOJ : 11/04/2011
DOL : 08/21/2013
Project name : Audit
DOJ : 09/11/2013
DOL : 09/01/2014
Project name : ContorlManagement
DOJ : 01/08/2015
DOL : 02/14/2016
Emp name is Cassey lives in Newyork
DOB : 03/09/1958
Project name : Healthcare
DOJ : 06/04/2011
DOL : 09/21/2011
Project name : ContorlManagement
DOJ : 01/08/2015
DOL : 02/14/2016
Emp name is Ronney lives in Alasca
DOB : 03/09/1958
Project name : Audit
DOJ : 09/11/2013
DOL : 09/01/2014
Project name : ContorlManagement
DOJ : 01/08/2015
DOL : 02/14/2016

这是我的代码：

val inp =  sc.textFile("F:\\Softwares\\Spark\\Programs for td\\input\\sample.txt").collect().mkString(" ").replaceAll("""[\r\n]+""", " ")
     val pt1 = """(?s)Dept : .*?(?=\bDept : |$)""".r
     val pt2 = """(?s)Emp name : .*?(?=\bEmp name : |$)""".r
     val dname = """Dept : (\S+).*?""".r
     val emprec = """Emp name : (\S+) lives in (\S+) DOB : (\S+).*?""".r
     val prrec = """Project name : (\S+) DOJ : (\S+) DOL : (\S+) """.r
     var recs = pt1.findAllIn(inp)
     var rec1 = recs

在这个 rec1(Regex.MatchIterator) 数据变成如下所示之后：

Dept : HR Emp name : Andrew lives in Colorodo DOB : 03/09/1958 Project name :Healthcare DOJ : 06/04/2011 DOL : 09/21/2011 Project name : Retail DOJ : 11/04/2011 DOL : 08/21/2013 Project name : Audit DOJ : 09/11/2013 DOL : 09/01/2014 Project name : ContorlManagement DOJ : 01/08/2015 DOL : 02/14/2016 Emp name : Alex lives in Texas DOB : 03/09/1958 Project name : Healthcare DOJ : 06/04/2011 DOL : 09/21/2011 Project name : ContorlManagement DOJ : 01/08/2015 DOL : 02/14/2016 Emp name : Mathew lives in California DOB : 03/09/1958 Project name : Healthcare DOJ : 06/04/2011 DOL : 09/21/2011 Project name : Retail DOJ : 11/04/2011 DOL : 08/21/2013 Project name : Audit DOJ : 09/11/2013 DOL : 09/01/2014 Project name : ContorlManagement DOJ : 01/08/2015 DOL : 02/14/2016  



Dept : QC Emp name : Nguyen lives in Nevada DOB : 03/09/1958 Project name : Healthcare DOJ : 06/04/2011 DOL : 09/21/2011 Project name : Retail DOJ : 11/04/2011 DOL : 08/21/2013 Project name : Audit DOJ : 09/11/2013 DOL : 09/01/2014 Project name : ContorlManagement DOJ : 01/08/2015 DOL : 02/14/2016 Emp name : Cassey lives in Newyork DOB : 03/09/1958 Project name : Healthcare DOJ : 06/04/2011 DOL : 09/21/2011 Project name : ContorlManagement DOJ : 01/08/2015 DOL : 02/14/2016 Emp name : Ronney lives in Alasca DOB : 03/09/1958 Project name : Audit DOJ : 09/11/2013 DOL : 09/01/2014 Project name : ContorlManagement DOJ : 01/08/2015 DOL : 02/14/2016

即每个 Dept 数据成为它自己的行。有时它可能在 300k - 400k 行之间变化。这是代码的下一部分，它对其进行格式化并将它们转换为数据帧。（我还没有编写代码将这个 Df 插入配置单元。但我认为，我可以在这个 pgm 的底部写这个。）

   var recfinal =  recs.map
      {
        e=>
 //  println(e)
      var dname(d) = e
   //    println(d)
      var rec2= pt2.findAllIn(e)

        var rec5 =rec2.map { k =>
    //   println(k)
                var emprec(ename,est,edob) = k
     //     println(ename,est,edob)
                var rec3 = prrec.findAllIn(k) 

               var rec4 =  rec3.map { j =>
                            var prrec(prjname,doj,dol) = j
                            var reclist = (d,ename,est,edob,prjname,doj,dol)
                             reclist
                           }
                .toList
                rec4
                  }
       .toList
       rec5

      }
    .flatMap(identity)
   .flatMap(identity)
  val rec9= recfinal.toSeq.toDF("dname","en","st","db","prj","dj","dl")
  rec9.show

1）如何将代码的第二部分传播到多个执行器节点，同时数据也应该等价地拆分以被执行。就像我有 4 个节点集群，每个节点应该有 100K 行并单独执行第二个代码。但最后，这一切都应该在一张蜂巢表中。

4

0 回答 0