Iterating Collection In Two Threads

        from ord in sampleOrders
        group ord by ord.Product into orderGroups
        from grp in orderGroups
        select grp

sampleOrders.GroupBy(ord => ord.Product)
    .Select(grp => grp); // grp is an `IGrouping<,>`

sampleOrders.GroupBy(ord => ord.Product)
    .SelectMany(orderGroup => orderGroup); // orderGroup is an `IGrouping<,>`
    .Select(grp => grp); // grp is an Order

from x1 in e1
from x2 in e2
select v

( e1 ) . SelectMany( x1 => e2 , ( x1 , x2 ) => v )

from a in x
from b in a
select b

var orderGroups =              // orderGroups is plural: IEnumerable<IGrouping<,>>
    from ord in sampleOrders
    group ord by ord.Product;
var o =
    from orderGroup in orderGroups // orderGroup is singular: IGrouping<,>
    from item in orderGroup        // `item` instead of `grp`
    select item;

var o = 
        from ord in sampleOrders
        group ord by ord.Product into orderGroup
        from item in orderGroup
        select item;               // item is an order, not a group of orders

import threading
from multiprocessing.pool import ThreadPool

def test_thread(elem):
    return elem ** 2

a = [1,2,3,4,5,6,7,8]
pool = ThreadPool(2) # 2 worker threads
results = []
for x in range(8):
    print x
    results.append(pool.apply_async(test_thread, args=(a[x],)))

results = [result.get() for result in results]
# You can also replace this for loop altogether using pool.map
# and get the same result:
# results = pool.map(test_thread, range(8))
print(results)

0
1
2
3
4
5
6
7
[1, 4, 9, 16, 25, 36, 49, 64]

val records: RDD[List[String]] = ??? //whatever data
var list:mutable.List[String] = List()
for {record <- records
     entry <- records } 
    { list += entry }

records.foreach{ record =>     //RDD.foreach => serializes closure and executes remotely
     record.foreach{entry =>   //record.foreach => local operation on the record collection
        list += entry          // this mutable list object is updated in each executor but never sent back to the driver. All updates are lost  
     }
}

def compareFields(colName:String, row1:Row, row2:Row): Option[DiscrepancyData] = {
    val key = "year"
    val v1 = row1.getAs(colName).toString
    val v2 = row2.getAs(colName).toString
    if (v1 != v2){
        Some(DiscrepancyData(
            row1.getAs(key).toString, //fieldKey
            colName, //fieldName
            v1, //table1Value
            v2, //table2Value
            v2) //expectedValue
        )
    } else None
}

val discrepancies = table.flatMap{case (str, row) =>
    compareCols.flatMap{col => compareFields(col, row.next, row.next) }   
}

val discrepancies = for {
    (str,row) <- table
    col <- compareCols
    dis <- compareFields(col, row.next, row.next)
} yield dis

val materializedDiscrepancies = discrepancies.collect()