商品名称 单价 进价 华为手机 4000 3500 Hp电脑 8000 7500 利用MapReduce编程框架编写程序统计每种商品的销售情况，统计出利润最高和销量最好的商品。

假设销售数据已经按照一定格式存储在一个大型文本文件中，每一行表示一次销售记录，包含商品名称、销售价格和进货价格。可以按照以下步骤完成利润和销量的统计：

1. Map阶段：将每一行数据按照商品名称作为Key，将销售价格和进货价格作为Value，输出<Key, Value>对。

2. Reduce阶段：对于每一个Key，将其对应的所有记录的Value相加，得到该商品的总销售额和总进货成本。利用这两个值计算该商品的总利润和总销量。

3. 再次Map阶段：将每一个商品的名称作为Key，将其销售额和销量作为Value，输出<Key, Value>对。

4. 再次Reduce阶段：对于每一个Key，将其对应的所有记录的Value相加，得到该商品的总销售额和总销量。根据这两个值可以计算该商品的平均销售价格和平均进货价格，以及利润率和销售量。

5. 根据统计结果，选出利润最高和销量最好的商品。

示例代码如下（假设输入文件路径为input.txt，输出文件路径为output.txt）：

from mrjob.job import MRJob
from mrjob.step import MRStep

class SalesAnalysis(MRJob):
    
    def steps(self):
        return [
            MRStep(mapper=self.mapper_profit,
                   reducer=self.reducer_profit),
            MRStep(mapper=self.mapper_sales,
                   reducer=self.reducer_sales),
            MRStep(mapper=self.mapper_summary,
                   reducer=self.reducer_summary),
            MRStep(mapper=self.mapper_top,
                   reducer=self.reducer_top)
        ]
    
    def mapper_profit(self, _, line):
        fields = line.strip().split()
        product = fields[0]
        sales = float(fields[1])
        cost = float(fields[2])
        yield product, (sales, cost)
    
    def reducer_profit(self, product, values):
        total_sales = 0.0
        total_cost = 0.0
        for sale, cost in values:
            total_sales += sale
            total_cost += cost
        profit = total_sales - total_cost
        yield product, (total_sales, total_cost, profit)
    
    def mapper_sales(self, product, values):
        total_sales, total_cost, profit = values
        yield product, (total_sales, 1)
    
    def reducer_sales(self, product, values):
        total_sales = 0.0
        total_count = 0
        for sale, count in values:
            total_sales += sale
            total_count += count
        avg_price = total_sales / total_count
        yield product, (total_sales, total_count, avg_price)
    
    def mapper_summary(self, product, values):
        total_sales, total_count, avg_price = values
        yield None, (product, total_sales, total_count, avg_price)
    
    def reducer_summary(self, _, values):
        top_profit = ('', 0.0)
        top_sales = ('', 0)
        for product, total_sales, total_count, avg_price in values:
            profit_rate = total_sales / avg_price
            if profit_rate > top_profit[1]:
                top_profit = (product, profit_rate)
            if total_count > top_sales[1]:
                top_sales = (product, total_count)
        yield 'profit', top_profit[0]
        yield 'sales', top_sales[0]
    
    def mapper_top(self, category, product):
        yield category, product
    
    def reducer_top(self, category, products):
        top_product = max(products)
        yield category, top_product
    
if __name__ == '__main__':
    SalesAnalysis.run()

在命令行中执行以下命令即可运行程序：

python sales_analysis.py input.txt -o output.txt

输出文件output.txt中会包含利润最高和销量最好的商品名称。